Update 02_ad_model_dnn_v11_update: add pai flow

ad/02_ad_model_dnn_v11_update.sh

@@ -215,31 +215,13 @@ bucket_feature_from_origin_to_hive() {
   )
 }
 
-xgb_train() {
+run_pai_flow() {
   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_20250104 \
-  --master yarn --driver-memory 6G --executor-memory 10G --executor-cores 2 --num-executors 11 \
-  --conf spark.yarn.executor.memoryoverhead=2048 \
-  --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/fengzhoutian/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:10 repartition:20 \
-  negSampleRate:0.04
-
+  python ad/pai_flow_operator2.py
+  
   local return_code=$?
-  check_run_status ${return_code} ${step_start_time} "XGB模型训练任务" "XGB模型训练失败"
+  check_run_status ${return_code} ${step_start_time} "PAI工作流任务" "PAI工作流执行失败"
 }
 
 calc_model_predict() {
@@ -361,48 +343,6 @@ draw_q_distribution() {
   python_return_code=$?
 }
 
-
-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下载模型任务" "HDFS下载模型失败" 
-      exit 1 
-    fi
-
-    tar -czvf ${model_name}_tmp.tar.gz -C ${model_name} .
-    
-    # 将模型文件和校准文件推送到OSS上
-    ${HADOOP} fs -put -f ${model_name}_tmp.tar.gz ${MODEL_OSS_PATH}
-    local return_code=$?
-    check_run_status ${return_code} ${step_start_time} "模型上传OSS任务" "模型上传OSS失败"
-
-    ${HADOOP} fs -cp -f ${MODEL_OSS_PATH}/${model_name}_tmp.tar.gz ${MODEL_OSS_PATH}/${model_name}.tar.gz
-    local return_code=$?
-    check_run_status ${return_code} ${step_start_time} "模型上传OSS任务" "模型重命名OSS失败"
-
-    echo ${model_save_path} > ${model_path_file}
-  )
-
-  local return_code=$?
-  check_run_status ${return_code} ${step_start_time} "模型上传OSS任务" "模型上传OSS失败"
-
-  local step_end_time=$(date +%s)
-  local elapsed=$((${step_end_time} - ${start_time}))
-  echo -e "${LOG_PREFIX} -- 模型更新完成 -- 模型更新成功: 耗时 ${elapsed}"
-  
-  send_success_upload_msg
-  # 
-  rm -f ./${model_name}.tar.gz
-  rm -rf ./${model_name}
-  rm -rf ${OSS_CALIBRATION_FILE_NAME}.txt
-}
-
 make_train_node_conf() {
   train_data_path=''
   TABLE_PART_PREFIX="odps://loghubods/tables/ad_easyrec_train_data_v3_sampled/dt"
@@ -444,18 +384,8 @@ main() {
   check_ad_hive
   bucket_feature_from_origin_to_hive
   make_train_node_conf
-  exit
-
-  if [ "${current_day_of_week}" -eq 1 ] || [ "${current_day_of_week}" -eq 3 ] || [ "${current_day_of_week}" -eq 5 ]; then
-    echo "当前是周一，周三或周五，开始训练并更新模型"
-    xgb_train
-    model_predict
-    compare_predictions
-    draw_q_distribution
-    model_upload_oss
-  else
-    echo "当前是周一，周三或周五，不更新模型"
-  fi 
+
+  run_pai_flow
 }