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

qiaojialiang/handle_rov.sh

@@ -5,6 +5,8 @@ set -ex
 
 # 原始数据table name
 table='alg_recsys_sample_all'
+today="$(date +%Y%m%d)"
+today_early_3="$(date -d '3 days ago' +%Y%m%d)"
 #table='alg_recsys_sample_all_test'
 # 处理分区配置 推荐数据间隔一天生产，所以5日0点使用3日0-23点数据生产new模型数据
 begin_early_2_Str="$(date -d '2 days ago' +%Y%m%d)"
@@ -15,77 +17,123 @@ endHhStr=23
 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_name=akaqjl8
+FM_HOME=/root/sunmingze/alphaFM
 
-# 0 判断上游表是否生产完成，最长等待到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
 
-# 1 生产原始数据
-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}
+## 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
+#
+## 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原始样本生产任务执行失败"
+#   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 模型训练
+#$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 -core 8
+#if [ $? -ne 0 ]; then
+#   echo "模型训练失败"
+#   /root/anaconda3/bin/python ad/ad_monitor_util.py "推荐模型训练失败"
+#   exit 1
+#fi
+
+$HADOOP fs -text ${bucketDataPath}/20240703/* | ${FM_HOME}/fm_train -m ${MODEL_PATH}/${model_name}_20240703.txt -dim 1,1,8 -core 8
 if [ $? -ne 0 ]; then
-   echo "Spark原始样本生产任务执行失败"
+   echo "模型训练失败"
+   /root/anaconda3/bin/python ad/ad_monitor_util.py "推荐模型训练失败"
    exit 1
-else
-    echo "spark原始样本生产执行成功"
 fi
 
 
-# 2 特征值拼接
-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
 
-# 3 特征分桶
-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