推荐模型自动化更新-单次模型训练

qiaojialiang/handle_rov.sh

@@ -1,6 +1,8 @@
 #!/bin/sh
 set -ex
 
+source /root/anaconda3/bin/activate py37
+
 #  nohup sh handle_rov.sh > "$(date +%Y%m%d_%H%M%S)_handle_rov.log" 2>&1 &
 
 # 原始数据table name
@@ -13,129 +15,148 @@ begin_early_2_Str="$(date -d '2 days ago' +%Y%m%d)"
 end_early_2_Str="$(date -d '2 days ago' +%Y%m%d)"
 beginHhStr=00
 endHhStr=23
+max_hour=05
+max_minute=00
 # 各节点产出hdfs文件绝对路径
+# 源数据文件
 originDataPath=/dw/recommend/model/13_sample_data/
+# 特征值
 valueDataPath=/dw/recommend/model/14_feature_data/
+# 特征分桶
 bucketDataPath=/dw/recommend/model/16_train_data/
-MODEL_PATH=/root/joe/recommend-emr-dataprocess/rov/model
-PREDICT_PATH=/root/zhaohp/recommend-emr-dataprocess/predict
+# 模型数据路径
+MODEL_PATH=/root/joe/recommend-emr-dataprocess/model
+# 预测路径
+PREDICT_PATH=/root/joe/recommend-emr-dataprocess/predict
+# 历史线上正在使用的模型数据路径
 LAST_MODEL_HOME=/root/joe/model_online
+# 模型数据文件前缀
 model_name=akaqjl8
+# fm模型
 FM_HOME=/root/sunmingze/alphaFM/bin
+# hadoop
 HADOOP=/opt/apps/HADOOP-COMMON/hadoop-common-current/bin/hadoop
 
 
-## 0 对比AUC 前置对比2日模型数据 与 线上模型数据效果对比，如果2日模型优于线上，更新线上模型
-#online_model=${MODEL_PATH}/model_online.txt
-#$HADOOP fs -text ${bucketDataPath}/${today}/* | /root/sunmingze/alphaFM/bin/fm_predict -m ${MODEL_PATH}/${online_model} -dim 0 -core 8 -out ${PREDICT_PATH}/${model_name}_${today}_online.txt
-#$HADOOP fs -text ${bucketDataPath}/${today}/* | /root/sunmingze/alphaFM/bin/fm_predict -m ${MODEL_PATH}/${model_name}_${today_early_3}.txt -dim 0 -core 8 -out ${PREDICT_PATH}/${model_name}_${today}_new.txt
+
+# 0 对比AUC 前置对比2日模型数据 与 线上模型数据效果对比，如果2日模型优于线上，更新线上模型
+#echo "$(date +%Y-%m-%d_%H-%M-%S)----------step0------------开始对比，新：${MODEL_PATH}/${model_name}_${today_early_3}.txt，与线上online模型数据auc效果"
+#$HADOOP fs -text ${bucketDataPath}/${begin_early_2_Str}/* | ${FM_HOME}/bin/fm_predict -m ${LAST_MODEL_HOME}/model_online.txt -dim 8 -core 8 -out ${PREDICT_PATH}/${model_name}_${today}_online.txt
+#$HADOOP fs -text ${bucketDataPath}/${begin_early_2_Str}/* | ${FM_HOME}/bin/fm_predict -m ${MODEL_PATH}/${model_name}_${today_early_3}.txt -dim 8 -core 8 -out ${PREDICT_PATH}/${model_name}_${today}_new.txt
 #
-## 1 对比auc数据判断是否更新线上模型
 #online_auc=`cat ${PREDICT_PATH}/${model_name}_${today}_online.txt | /root/sunmingze/AUC/AUC`
-#new_auc=`cat ${PREDICT_PATH}/${model_name}_${today}_new.txt | /root/sunmingze/AUC/AUC`
-#if [ "$online_auc" \< "$new_auc" ]; then
-#    echo "推荐新模型优于线上模型: 线上模型AUC: ${online_auc}, 新模型AUC: ${new_auc}"
-#    /root/anaconda3/bin/python ad/ad_monitor_util.py "新模型优于线上模型: 线上模型AUC: ${online_auc}, 新模型AUC: ${new_auc}"
-#    # todo 模型格式转换
-#
-#    # todo 模型文件上传OSS
-#
-#    # todo 本地保存最新的线上使用的模型，用于下一次的AUC验证
-#else
-#    echo "推荐新模型不如线上模型: 线上模型AUC: ${online_auc}, 新模型AUC: ${new_auc}"
-#    /root/anaconda3/bin/python ad/ad_monitor_util.py "新模型不如线上模型: 线上模型AUC: ${online_auc}, 新模型AUC: ${new_auc}"
-##    exit 1
-#fi
-#
-#
-## 2 判断上游表是否生产完成，最长等待到12点
-## shellcheck disable=SC2039
-#source /root/anaconda3/bin/activate py37
-## shellcheck disable=SC2154
-#echo "$(date +%Y-%m-%d_%H-%M-%S)----------step1------------开始校验是否生产完数据，分区信息：beginStr:${begin_early_2_Str}${beginHhStr},endStr:${end_early_2_Str}${endHhStr}"
-#while true; do
-#  python_return_code=$(python /root/joe/recommend-emr-dataprocess/qiaojialiang/checkHiveDataUtil.py --table ${table} --beginStr ${begin_early_2_Str}${beginHhStr} --endStr ${end_early_2_Str}${endHhStr})
-#  echo "python 返回值：${python_return_code}"
-#  if [ $python_return_code -eq 0 ]; then
-#    echo "Python程序返回0，校验存在数据，退出循环。"
-#    break
-#  fi
-#  echo "Python程序返回非0值，不存在数据，等待五分钟后再次调用。"
-#  sleep 300
-#  current_hour=$(date +%H)
-#  current_minute=$(date +%M)
-#  # shellcheck disable=SC2039
-#  if (( current_hour > max_hour || (current_hour == max_hour && current_minute >= max_minute) )); then
-#    echo "最长等待时间已到，失败:${current_hour}-${current_minute}"
-#    exit 1
-#  fi
-#done
-#
-## 3 生产原始数据
-#echo "$(date +%Y-%m-%d_%H-%M-%S)----------step2------------开始根据${table}生产原始数据"
-#/opt/apps/SPARK2/spark-2.4.8-hadoop3.2-1.0.8/bin/spark-class2 org.apache.spark.deploy.SparkSubmit \
-#--class com.aliyun.odps.spark.examples.makedata_qiao.makedata_13_originData_20240705 \
-#--master yarn --driver-memory 1G --executor-memory 2G --executor-cores 1 --num-executors 16 \
-#../target/spark-examples-1.0.0-SNAPSHOT-shaded.jar \
-#tablePart:64 repartition:32 \
-#beginStr:${begin_early_2_Str}${beginHhStr} endStr:${end_early_2_Str}${endHhStr} \
-#savePath:${originDataPath} \
-#table:${table}
 #if [ $? -ne 0 ]; then
-#   echo "Spark原始样本生产任务执行失败"
+#   echo "推荐线上模型AUC计算失败"
+##   /root/anaconda3/bin/python ad/ad_monitor_util.py "线上模型AUC计算失败"
 #   exit 1
-#else
-#    echo "spark原始样本生产执行成功"
 #fi
 #
-#
-## 4 特征值拼接
-#echo "$(date +%Y-%m-%d_%H-%M-%S)----------step3------------开始特征值拼接"
-#/opt/apps/SPARK2/spark-2.4.8-hadoop3.2-1.0.8/bin/spark-class2 org.apache.spark.deploy.SparkSubmit \
-#--class com.aliyun.odps.spark.examples.makedata_qiao.makedata_14_valueData_20240705 \
-#--master yarn --driver-memory 1G --executor-memory 3G --executor-cores 1 --num-executors 32 \
-#../target/spark-examples-1.0.0-SNAPSHOT-shaded.jar \
-#readPath:${originDataPath} \
-#savePath:${valueDataPath} \
-#beginStr:${begin_early_2_Str} endStr:${end_early_2_Str} repartition:1000
+#new_auc=`${PREDICT_PATH}/${model_name}_${today}_new.txt | /root/sunmingze/AUC/AUC`
 #if [ $? -ne 0 ]; then
-#   echo "Spark特征值拼接处理任务执行失败"
+#   echo "推荐新模型AUC计算失败"
+##   /root/anaconda3/bin/python ad/ad_monitor_util.py "新模型AUC计算失败"
 #   exit 1
-#else
-#   echo "spark特征值拼接处理执行成功"
 #fi
 #
-## 5 特征分桶
-#echo "$(date +%Y-%m-%d_%H-%M-%S)----------step4------------根据特征分桶生产重打分特征数据"
-#/opt/apps/SPARK2/spark-2.4.8-hadoop3.2-1.0.8/bin/spark-class2 org.apache.spark.deploy.SparkSubmit \
-#--class com.aliyun.odps.spark.examples.makedata_qiao.makedata_16_bucketData_20240705 \
-#--master yarn --driver-memory 2G --executor-memory 4G --executor-cores 1 --num-executors 16 \
-#../target/spark-examples-1.0.0-SNAPSHOT-shaded.jar \
-#readPath:${valueDataPath} \
-#savePath:${bucketDataPath} \
-#beginStr:${begin_early_2_Str} endStr:${end_early_2_Str} repartition:1000
-#if [ $? -ne 0 ]; then
-#   echo "Spark特征分桶处理任务执行失败"
-#   exit 1
-#else
-#   echo "spark特征分桶处理执行成功"
-#fi
 #
-## 6 模型训练
-#$HADOOP fs -text ${bucketDataPath}/${begin_early_2_Str}/* | ${FM_HOME}/fm_train -m ${MODEL_PATH}/${model_name}_${begin_early_2_Str}.txt -dim 1,1,8 -im ${LAST_MODEL_HOME}/model_online.txt -core 8
-#if [ $? -ne 0 ]; then
-#   echo "模型训练失败"
-#   /root/anaconda3/bin/python ad/ad_monitor_util.py "推荐模型训练失败"
-#   exit 1
+## 1 对比auc数据判断是否更新线上模型
+#if [ "$online_auc" \< "$new_auc" ]; then
+#    echo "新模型优于线上模型: 线上模型AUC: ${online_auc}, 新模型AUC: ${new_auc}"
+#    # todo 模型格式转换
+#
+#    # todo 模型文件上传OSS
+#
+#    # todo 本地保存最新的线上使用的模型，用于下一次的AUC验证
+##    /root/anaconda3/bin/python ad/ad_monitor_util.py "新模型优于线上模型: 线上模型AUC: ${online_auc}, 新模型AUC: ${new_auc}"
+#else
+#    echo "新模型不如线上模型: 线上模型AUC: ${online_auc}, 新模型AUC: ${new_auc}"
+##    /root/anaconda3/bin/python ad/ad_monitor_util.py "新模型不如线上模型: 线上模型AUC: ${online_auc}, 新模型AUC: ${new_auc}"
 #fi
 
-$HADOOP fs -text ${bucketDataPath}/20240703/* | ${FM_HOME}/fm_train -m ${MODEL_PATH}/${model_name}_20240703.txt -dim 1,1,8 -im ${LAST_MODEL_HOME}/model_online.txt -core 8
+
+
+# 2 判断上游表是否生产完成，最长等待到12点
+# shellcheck disable=SC2154
+echo "$(date +%Y-%m-%d_%H-%M-%S)----------step1------------开始校验是否生产完数据，分区信息：beginStr:${begin_early_2_Str}${beginHhStr},endStr:${end_early_2_Str}${endHhStr}"
+while true; do
+  python_return_code=$(python /root/joe/recommend-emr-dataprocess/qiaojialiang/checkHiveDataUtil.py --table ${table} --beginStr ${begin_early_2_Str}${beginHhStr} --endStr ${end_early_2_Str}${endHhStr})
+  echo "python 返回值：${python_return_code}"
+  if [ $python_return_code -eq 0 ]; then
+    echo "Python程序返回0，校验存在数据，退出循环。"
+    break
+  fi
+  echo "Python程序返回非0值，不存在数据，等待五分钟后再次调用。"
+  sleep 300
+  current_hour=$(date +%H)
+  current_minute=$(date +%M)
+  # shellcheck disable=SC2039
+  if (( current_hour > max_hour || (current_hour == max_hour && current_minute >= max_minute) )); then
+    echo "最长等待时间已到，失败:${current_hour}-${current_minute}"
+    exit 1
+  fi
+done
+
+# 3 生产原始数据
+echo "$(date +%Y-%m-%d_%H-%M-%S)----------step2------------开始根据${table}生产原始数据"
+/opt/apps/SPARK2/spark-2.4.8-hadoop3.2-1.0.8/bin/spark-class2 org.apache.spark.deploy.SparkSubmit \
+--class com.aliyun.odps.spark.examples.makedata_qiao.makedata_13_originData_20240705 \
+--master yarn --driver-memory 1G --executor-memory 2G --executor-cores 1 --num-executors 16 \
+../target/spark-examples-1.0.0-SNAPSHOT-shaded.jar \
+tablePart:64 repartition:32 \
+beginStr:${begin_early_2_Str}${beginHhStr} endStr:${end_early_2_Str}${endHhStr} \
+savePath:${originDataPath} \
+table:${table}
+if [ $? -ne 0 ]; then
+   echo "Spark原始样本生产任务执行失败"
+   exit 1
+else
+   echo "spark原始样本生产执行成功"
+fi
+
+
+# 4 特征值拼接
+echo "$(date +%Y-%m-%d_%H-%M-%S)----------step3------------开始特征值拼接"
+/opt/apps/SPARK2/spark-2.4.8-hadoop3.2-1.0.8/bin/spark-class2 org.apache.spark.deploy.SparkSubmit \
+--class com.aliyun.odps.spark.examples.makedata_qiao.makedata_14_valueData_20240705 \
+--master yarn --driver-memory 1G --executor-memory 3G --executor-cores 1 --num-executors 32 \
+../target/spark-examples-1.0.0-SNAPSHOT-shaded.jar \
+readPath:${originDataPath} \
+savePath:${valueDataPath} \
+beginStr:${begin_early_2_Str} endStr:${end_early_2_Str} repartition:1000
+if [ $? -ne 0 ]; then
+   echo "Spark特征值拼接处理任务执行失败"
+   exit 1
+else
+   echo "spark特征值拼接处理执行成功"
+fi
+
+# 5 特征分桶
+echo "$(date +%Y-%m-%d_%H-%M-%S)----------step4------------根据特征分桶生产重打分特征数据"
+/opt/apps/SPARK2/spark-2.4.8-hadoop3.2-1.0.8/bin/spark-class2 org.apache.spark.deploy.SparkSubmit \
+--class com.aliyun.odps.spark.examples.makedata_qiao.makedata_16_bucketData_20240705 \
+--master yarn --driver-memory 2G --executor-memory 4G --executor-cores 1 --num-executors 16 \
+../target/spark-examples-1.0.0-SNAPSHOT-shaded.jar \
+readPath:${valueDataPath} \
+savePath:${bucketDataPath} \
+beginStr:${begin_early_2_Str} endStr:${end_early_2_Str} repartition:1000
+if [ $? -ne 0 ]; then
+   echo "Spark特征分桶处理任务执行失败"
+   exit 1
+else
+   echo "spark特征分桶处理执行成功"
+fi
+
+# 6 模型训练
+echo "$(date +%Y-%m-%d_%H-%M-%S)----------step5------------开始模型训练"
+$HADOOP fs -text ${bucketDataPath}/${begin_early_2_Str}/* | ${FM_HOME}/fm_train -m ${MODEL_PATH}/${model_name}_${begin_early_2_Str}.txt -dim 1,1,8 -im ${LAST_MODEL_HOME}/model_online.txt -core 8
 if [ $? -ne 0 ]; then
    echo "模型训练失败"
-   /root/anaconda3/bin/python ad/ad_monitor_util.py "推荐模型训练失败"
+#   /root/anaconda3/bin/python ad/ad_monitor_util.py "推荐模型训练失败"
    exit 1
 fi
 
+echo "$(date +%Y-%m-%d_%H-%M-%S)----------step6------------模型训练完成：${MODEL_PATH}/${model_name}_${begin_early_2_Str}.txt"
+