recommend-emr-dataprocess/qiaojialiang/handle_rov.sh

#!/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