recommend-emr-dataprocess/ad/01_ad_model_update.sh

#!/bin/sh
set -x

export PATH=$SPARK_HOME/bin:$PATH
export HADOOP_CONF_DIR=/etc/taihao-apps/hadoop-conf
export JAVA_HOME=/usr/lib/jvm/java-1.8.0

sh_path=$(cd $(dirname $0); pwd)
source ${sh_path}/00_common.sh

source /root/anaconda3/bin/activate py37


# 全局常量
HADOOP=/opt/apps/HADOOP-COMMON/hadoop-common-current/bin/hadoop
TRAIN_PATH=/dw/recommend/model/31_ad_sample_data_v4
BUCKET_FEATURE_PATH=/dw/recommend/model/33_ad_train_data_v4
TABLE=alg_recsys_ad_sample_all
# 特征文件名
feature_file=20240703_ad_feature_name.txt
# 模型本地临时保存路径
model_local_home=/root/zhaohp/XGB/

# 模型HDFS保存路径，测试时修改为其他路径，避免影响线上
MODEL_PATH=/dw/recommend/model/35_ad_model
# 预测结果保存路径，测试时修改为其他路径，避免影响线上
PREDICT_RESULT_SAVE_PATH=/dw/recommend/model/34_ad_predict_data
# 模型OSS保存路径，测试时修改为其他路径，避免影响线上
MODEL_OSS_PATH=oss://art-recommend.oss-cn-hangzhou.aliyuncs.com/zhangbo/
# 线上模型名，测试时修改为其他模型名，避免影响线上
model_name=model_xgb_351_1000_v2
# 本地保存HDFS模型路径文件，测试时修改为其他模型名，避免影响线上
model_path_file=${model_local_home}/online_model_path.txt

# 任务开始时间
start_time=$(date +%s)
# 前一天
today_early_1="$(date -d '1 days ago' +%Y%m%d)"
# 线上模型在HDFS中的路径
online_model_path=`cat ${model_path_file}`
# 训练用的数据路径
train_data_path=""
# 评估用的数据路径
predict_date_path=""
#评估结果保存路径
new_model_predict_result_path=""
# 模型保存路径
model_save_path=""
# 评测结果保存路径，后续需要根据此文件评估是否要更新模型
predict_analyse_file_path=""
# 保存模型评估的分析结果
old_incr_rate_avg=0
new_incr_rate_avg=0
old_incr_rate_list=""
new_incr_rate_list=""

# 校验命令的退出码
check_run_status() {
    local status=$1
    local step_start_time=$2
    local step_name=$3

    local step_end_time=$(date +%s)
    local step_elapsed=$(($step_end_time - $step_start_time))

    if [ $status -ne 0 ]; then
        echo "$LOG_PREFIX -- ${step_name}失败: 耗时 $step_elapsed"
        local elapsed=$(($step_end_time - $start_time))
        /root/anaconda3/bin/python ${sh_path}/ad_monitor_util.py --level error --msg "$msg" --start "$start_time" --elapsed "$elapsed"
        exit 1
    else
        echo "$LOG_PREFIX -- ${step_name}成功: 耗时 $step_elapsed"
    fi
}

init() {
  
  declare -a date_keys=()
  local count=1
  local current_data="$(date -d '2 days ago' +%Y%m%d)"
  # 循环获取前 n 天的非节日日期
  while [[ $count -lt 7 ]]; do
    date_key=$(date -d "$current_data" +%Y%m%d)
    # 判断是否是节日，并拼接训练数据路径
    if [ $(is_not_holidays $date_key) -eq 1 ]; then

      # 将 date_key 放入数组
      date_keys+=("$date_key")

      if [[ -z ${train_data_path} ]]; then
        train_data_path="${BUCKET_FEATURE_PATH}/${date_key}"
      else
        train_data_path="${BUCKET_FEATURE_PATH}/${date_key},${train_data_path}"
      fi 
      count=$((count + 1))
    else
      echo "日期: ${date_key}是节日，跳过"
    fi
    current_data=$(date -d "$current_data -1 day" +%Y%m%d)
  done

  last_index=$((${#date_keys[@]} - 1))
  train_first_day=${date_keys[$last_index]}
  train_last_day=${date_keys[0]}

  model_save_path=${MODEL_PATH}/${model_name}_${train_first_day: -4}_${train_last_day: -4}
  predict_date_path=${BUCKET_FEATURE_PATH}/${today_early_1}
  new_model_predict_result_path=${PREDICT_RESULT_SAVE_PATH}/${today_early_1}_351_1000_${train_first_day: -4}_${train_last_day: -4}
  online_model_predict_result_path=${PREDICT_RESULT_SAVE_PATH}/${today_early_1}_351_1000_${online_model_path: -9}
  predict_analyse_file_path=${model_local_home}/predict_analyse_file/${today_early_1}_351_1000_analyse.txt

  echo "init param train_data_path: ${train_data_path}"
  echo "init param predict_date_path: ${predict_date_path}"
  echo "init param new_model_predict_result_path: ${new_model_predict_result_path}"
  echo "init param online_model_predict_result_path: ${online_model_predict_result_path}"
  echo "init param model_save_path: ${model_save_path}"
  echo "init param online_model_path: ${online_model_path}"
  echo "init param feature_file: ${feature_file}"
  echo "init param model_name: ${model_name}"
  echo "init param model_local_home: ${model_local_home}"
  echo "init param model_oss_path: ${MODEL_OSS_PATH}"
  echo "init param predict_analyse_file_path: ${predict_analyse_file_path}"

  echo "当前Python环境安装的Python版本: $(python --version)"
  echo "当前Python环境安装的三方包: $(python -m pip list)"
}

# 校验大数据任务是否执行完成
check_ad_hive() {
  local step_start_time=$(date +%s)
  local max_hour=05
  local max_minute=30
  local elapsed=0
  while true; do
      local python_return_code=$(python ${sh_path}/ad_utils.py --excute_program check_ad_origin_hive --partition ${today_early_1} --hh 23)

      elapsed=$(($(date +%s) - $step_start_time))
      if [ "$python_return_code" -eq 0 ]; then
          break
      fi
      echo "Python程序返回非0值，等待五分钟后再次调用。"
      sleep 300
      local current_hour=$(date +%H)
      local current_minute=$(date +%M)
      if (( current_hour > max_hour || (current_hour == max_hour && current_minute >= max_minute) )); then
          local msg="大数据数据生产校验失败, 分区: ${today_early_1}"
          echo -e "$LOG_PREFIX -- 大数据数据生产校验 -- ${msg}: 耗时 $elapsed"
          /root/anaconda3/bin/python ${sh_path}/ad_monitor_util.py --level error --msg "$msg" --start "$start_time" --elapsed "$elapsed"
          exit 1
      fi
  done
  echo "$LOG_PREFIX -- 大数据数据生产校验 -- 大数据数据生产校验通过: 耗时 $elapsed"
}

origin_data() {
  (
    source ${sh_path}/25_xgb_make_data_origin_bucket.sh
    make_origin_data
  )
}

bucket_feature() {
  (
    source ${sh_path}/25_xgb_make_data_origin_bucket.sh
    make_bucket_feature
  )
}

xgb_train() {
  local step_start_time=$(date +%s)

  /opt/apps/SPARK3/spark-3.3.1-hadoop3.2-1.0.5/bin/spark-class org.apache.spark.deploy.SparkSubmit \
  --class com.tzld.piaoquan.recommend.model.train_01_xgb_ad_20240808 \
  --master yarn --driver-memory 6G --executor-memory 9G --executor-cores 1 --num-executors 31 \
  --conf spark.yarn.executor.memoryoverhead=1000 \
  --conf spark.shuffle.service.enabled=true \
  --conf spark.shuffle.service.port=7337 \
  --conf spark.shuffle.consolidateFiles=true \
  --conf spark.shuffle.manager=sort \
  --conf spark.storage.memoryFraction=0.4 \
  --conf spark.shuffle.memoryFraction=0.5 \
  --conf spark.default.parallelism=200 \
  /root/zhangbo/recommend-model/recommend-model-produce/target/recommend-model-produce-jar-with-dependencies.jar \
  featureFile:20240703_ad_feature_name.txt \
  trainPath:${train_data_path} \
  testPath:${predict_date_path} \
  savePath:${new_model_predict_result_path} \
  modelPath:${model_save_path} \
  eta:0.01 gamma:0.0 max_depth:5 num_round:1000 num_worker:30 repartition:20

  local return_code=$?
  check_run_status $return_code $step_start_time "XGB模型训练任务"
}

calc_model_predict() {
  local count=0
  local max_line=10
  local old_total_diff=0
  local new_total_diff=0
  while read -r line && [ ${count} -lt ${max_line} ]; do

      # 使用 ! 取反判断，只有当行中不包含 "cid" 时才执行继续的逻辑
      if [[ "${line}" == *"cid"* ]]; then
          continue
      fi

      read -a numbers <<< "${line}"

      if [[ -z ${old_incr_rate_list} ]];then
          old_incr_rate_list="${numbers[6]}"
          new_incr_rate_list="${numbers[7]}"
      else
          old_incr_rate_list="${old_incr_rate_list};${numbers[6]}"
          new_incr_rate_list="${new_incr_rate_list};${numbers[7]}"
      fi 

      old_total_diff=$( echo "${old_total_diff} + ${numbers[6]}" | bc -l )
      new_total_diff=$( echo "${new_total_diff} + ${numbers[7]}" | bc -l )

      count=$((${count} + 1))

  done < "${predict_analyse_file_path}"

  local return_code=$?
  check_run_status $return_code $step_start_time "计算Top10差异"

  old_incr_rate_avg=$( echo "scale=6; ${old_total_diff} / ${count}" | bc -l )
  check_run_status $? $step_start_time "计算老模型Top10差异"


  new_incr_rate_avg=$( echo "scale=6; ${new_total_diff} / ${count}" | bc -l )
  check_run_status $? $step_start_time "计算新模型Top10差异"

  echo "老模型Top10差异平均值: ${old_incr_rate_avg}"
  echo "老模型Top10差异列表: ${old_incr_rate_list}"
  echo "新模型Top10差异平均值: ${new_incr_rate_avg}"
  echo "新模型Top10差异列表: ${new_incr_rate_list}"
}

model_predict() {

  # 线上模型评估最新的数据
  local step_start_time=$(date +%s)
  /opt/apps/SPARK3/spark-3.3.1-hadoop3.2-1.0.5/bin/spark-class org.apache.spark.deploy.SparkSubmit \
  --class com.tzld.piaoquan.recommend.model.pred_01_xgb_ad_hdfsfile_20240813 \
  --master yarn --driver-memory 1G --executor-memory 1G --executor-cores 1 --num-executors 30 \
  --conf spark.yarn.executor.memoryoverhead=1024 \
  --conf spark.shuffle.service.enabled=true \
  --conf spark.shuffle.service.port=7337 \
  --conf spark.shuffle.consolidateFiles=true \
  --conf spark.shuffle.manager=sort \
  --conf spark.storage.memoryFraction=0.4 \
  --conf spark.shuffle.memoryFraction=0.5 \
  --conf spark.default.parallelism=200 \
  /root/zhangbo/recommend-model/recommend-model-produce/target/recommend-model-produce-jar-with-dependencies.jar \
  featureFile:20240703_ad_feature_name.txt \
  testPath:${predict_date_path} \
  savePath:${online_model_predict_result_path} \
  modelPath:${online_model_path}

  local return_code=$?
  check_run_status $return_code $step_start_time "线上模型评估${predict_date_path: -8}的数据"

  # 结果分析
  local python_return_code=$(python ${sh_path}/model_predict_analyse.py -p ${online_model_predict_result_path} ${new_model_predict_result_path} -f ${predict_analyse_file_path})
  check_run_status $python_return_code $step_start_time "线上模型评估${predict_date_path: -8}的数据"

  calc_model_predict

  if (( $(echo "${new_incr_rate_avg} > 0.100000" | bc -l ) ));then 
    check_run_status 1 $step_start_time "线上模型评估${predict_date_path: -8}的数据，绝对误差大于0.1，请检查"
    echo "线上模型评估${predict_date_path: -8}的数据，绝对误差大于0.1，请检查"
    exit 1
  fi 
}

model_upload_oss() {
  local step_start_time=$(date +%s)

  (
    cd ${model_local_home}

    ${HADOOP} fs -get ${model_save_path} ${model_name}
    if [ ! -d ${model_name} ]; then
      echo "从HDFS下载模型失败"
      check_run_status 1 $step_start_time "HDFS下载模型任务" 
      exit 1 
    fi

    tar -czvf ${model_name}.tar.gz -C ${model_name} .

    rm -rf ${model_name}.tar.gz.crc

    ${HADOOP} fs -rm -r -skipTrash ${MODEL_OSS_PATH}/${model_name}.tar.gz
    
    ${HADOOP} fs -put ${model_name}.tar.gz ${MODEL_OSS_PATH}
    local return_code=$?
    check_run_status $return_code $step_start_time "模型上传OSS任务"


    echo ${model_save_path} > ${model_path_file}

    rm -f ./${model_name}.tar.gz
    rm -rf ./${model_name}
  )

  local return_code=$?
  check_run_status $return_code $step_start_time "模型上传OSS任务"

  local step_end_time=$(date +%s)
  local elapsed=$(($step_end_time - $start_time))

  echo -e "$LOG_PREFIX -- 模型更新完成 -- 模型更新成功: 耗时 $elapsed"
  local msg="\n\t - 广告模型文件更新完成 \n\t - 前一天线上模型全天Top差异平均值: ${old_incr_rate_avg} \n\t - 前一天线上模型全天Top差异: ${old_incr_rate_list} \n\t - 前一天新模型全天Top10差异平均值: ${new_incr_rate_avg} \n\t - 前一天新模型全天Top差异: ${new_incr_rate_list} \n\t - 模型上传路径: ${MODEL_OSS_PATH}/${model_name}.tar.gz"
  /root/anaconda3/bin/python ${sh_path}/ad_monitor_util.py --level info --msg "${msg}" --start "${start_time}" --elapsed "${elapsed}"
}

# 主方法
main() {
  init

  check_ad_hive

  origin_data

  bucket_feature

  xgb_train

  model_predict

  model_upload_oss

}


main