#!/bin/sh set -ex # nohup sh handle_rov.sh > "$(date +%Y%m%d_%H%M%S)_handle_rov.log" 2>&1 & # 原始数据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)" end_early_2_Str="$(date -d '2 days ago' +%Y%m%d)" beginHhStr=00 endHhStr=23 # 各节点产出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 LAST_MODEL_HOME=/root/joe/model_online model_name=akaqjl8 FM_HOME=/root/sunmingze/alphaFM ## 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 -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 #fi echo ${bucketDataPath}/20240703/* echo ${FM_HOME}/fm_train echo ${MODEL_PATH}/${model_name}_20240703.txt echo ${LAST_MODEL_HOME}/model_online.txt #$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 #if [ $? -ne 0 ]; then # echo "模型训练失败" # /root/anaconda3/bin/python ad/ad_monitor_util.py "推荐模型训练失败" # exit 1 #fi