feat:添加在特征分桶阶段修改特征值的逻辑

ad/20_new_ad__model_train_predict_auc.sh

@@ -1,6 +1,6 @@
 #!/bin/sh
 
-# 模型训练，预测，计算AUC脚本
+# 训练新模型，并使用后面的数据计算AUC，评估模型效果
 
 set -x
 
@@ -64,4 +64,7 @@ main() {
 
 }
 
-main
+main
+
+
+# nohup ./ad/20_new_ad__model_train_predict_auc.sh 20240712 20240717 model_bkb8_v4 8 > logs/20_ad_model_train_predict_auc.log 2>&1 &

ad/22_ad__model_add_dt_train_predict_auc.sh

@@ -0,0 +1,71 @@
+#!/bin/sh
+
+# 模型增量训练，预测，计算AUC脚本
+
+set -x
+
+begin_date=$1
+end_date=$2
+model_name=$3
+train_dim=$4
+predict_dim=$5
+
+PROJECT_HOME=/root/zhaohp/recommend-emr-dataprocess
+HADOOP=/opt/apps/HADOOP-COMMON/hadoop-common-current/bin/hadoop
+HDFS_TRAIN_DATE_PATH=/dw/recommend/model/33_ad_train_data_v4
+MODEL_PATH=${PROJECT_HOME}/model
+PREDICT_PATH=${PROJECT_HOME}/predict
+
+FM_TRAIN=/root/sunmingze/alphaFM/bin/fm_train
+FM_PREDICT=/root/sunmingze/alphaFM/bin/fm_predict
+
+train_date=$begin_date
+
+# 计算模型的AUC，从训练日期的后一天到参数的end_date
+predict_auc() {
+    echo -e "\t==================== 开始预测 $train_date 模型 ===================="
+
+    predict_date=$(date -d "$train_date +1 day" +%Y%m%d)
+    predict_end_date=$(date -d "$end_date +1 day" +%Y%m%d)
+    while [ "$predict_date" != "$predict_end_date" ]; do
+
+        $HADOOP fs -text ${HDFS_TRAIN_DATE_PATH}/${predict_date}/* | ${FM_PREDICT} -m ${MODEL_PATH}/${model_name}_${train_date}.txt -dim ${predict_dim} -core 8 -out ${PREDICT_PATH}/${model_name}_${train_date}.txt
+        auc=`cat ${PREDICT_PATH}/${model_name}_${train_date}.txt | /root/sunmingze/AUC/AUC`
+
+        echo "模型训练日期: ${train_date}, 模型预测日期: ${predict_date}, AUC: ${auc}, 模型路径: ${MODEL_PATH}/${model_name}_${train_date}.txt"
+
+        predict_date=$(date -d "$predict_date +1 day" +%Y%m%d)
+
+    done
+
+    echo -e "\n\t==================== 预测 $train_date 模型结束 ===================="
+
+}
+main() {
+
+    # 增量训练模型
+    while [ "$train_date" != "$end_date" ]; do
+        echo "==================== 开始训练 $train_date 模型 ===================="
+
+        # 模型训练
+        yesterday=$(date -d "$train_date -1 day" +%Y%m%d)
+
+        input_model=${MODEL_PATH}/${model_name}_${yesterday}.txt
+        if [ ! -e "${input_model}" ]; then
+            echo "输入模型: ${input_model} 不存在，退出"
+            exit 1
+        fi
+
+        $HADOOP fs -text ${HDFS_TRAIN_DATE_PATH}/${train_date}/* | ${FM_TRAIN} -m ${MODEL_PATH}/${model_name}_${train_date}.txt -dim ${train_dim} -core 8 -im ${input_model}
+
+        predict_auc
+
+        train_date=$(date -d "$train_date +1 day" +%Y%m%d)
+
+        echo "==================== 训练 $train_date 模型结束 ===================="
+        echo -e "\n\n\n\n\n\n"
+    done
+
+}
+
+main

src/main/scala/com/aliyun/odps/spark/examples/makedata_ad/makedata_ad_33_bucketData_default_value_20240718.scala

@@ -0,0 +1,140 @@
+package com.aliyun.odps.spark.examples.makedata_ad
+
+import com.alibaba.fastjson.JSON
+import com.aliyun.odps.spark.examples.myUtils.{MyDateUtils, MyHdfsUtils, ParamUtils}
+import examples.extractor.ExtractorUtils
+import org.apache.hadoop.io.compress.GzipCodec
+import org.apache.spark.sql.SparkSession
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ArrayBuffer
+import scala.io.Source
+/*
+
+ */
+
+object makedata_ad_33_bucketData_default_value_20240718 {
+  def main(args: Array[String]): Unit = {
+
+    val spark = SparkSession
+      .builder()
+      .appName(this.getClass.getName)
+      .getOrCreate()
+    val sc = spark.sparkContext
+
+    val loader = getClass.getClassLoader
+
+    val resourceUrlBucket = loader.getResource("20240718_ad_bucket_688.txt")
+    val buckets =
+      if (resourceUrlBucket != null) {
+        val buckets = Source.fromURL(resourceUrlBucket).getLines().mkString("\n")
+        Source.fromURL(resourceUrlBucket).close()
+        buckets
+      } else {
+        ""
+      }
+    println(buckets)
+    val bucketsMap = buckets.split("\n")
+      .map(r => r.replace(" ", "").replaceAll("\n", ""))
+      .filter(r => r.nonEmpty)
+      .map(r =>{
+        val rList = r.split("\t")
+        (rList(0), (rList(1).toDouble, rList(2).split(",").map(_.toDouble)))
+      }).toMap
+    val bucketsMap_br = sc.broadcast(bucketsMap)
+
+
+    // 1 读取参数
+    val param = ParamUtils.parseArgs(args)
+    val readPath = param.getOrElse("readPath", "/dw/recommend/model/31_ad_sample_data/")
+    val savePath = param.getOrElse("savePath", "/dw/recommend/model/33_ad_train_data/")
+    val beginStr = param.getOrElse("beginStr", "20240620")
+    val endStr = param.getOrElse("endStr", "20240620")
+    val repartition = param.getOrElse("repartition", "100").toInt
+    val filterNames = param.getOrElse("filterNames", "").split(",").toSet
+    val whatLabel = param.getOrElse("whatLabel", "ad_is_conversion")
+    val modifyFeatureName= param.getOrElse("modifyName", "").split(",").toSet
+    val defaultValue= param.getOrElse("defaultValue", "0.01")
+
+    val dateRange = MyDateUtils.getDateRange(beginStr, endStr)
+    for (date <- dateRange) {
+      println("开始执行:" + date)
+      val data = sc.textFile(readPath + "/" + date + "*").map(r=>{
+        val rList = r.split("\t")
+        val logKey = rList(0)
+        val labelKey = rList(1)
+        val jsons = JSON.parseObject(rList(2))
+        val features = scala.collection.mutable.Map[String, Double]()
+        jsons.foreach(r => {
+          features.put(r._1, jsons.getDoubleValue(r._1))
+        })
+        (logKey, labelKey, features)
+      })
+        .filter{
+          case (logKey, labelKey, features) =>
+            val logKeyList = logKey.split(",")
+            val apptype = logKeyList(0)
+            !Set("12", "13").contains(apptype)
+        }
+        .map{
+          case (logKey, labelKey, features) =>
+            val label = JSON.parseObject(labelKey).getOrDefault(whatLabel, "0").toString
+            (label, features)
+        }
+        .mapPartitions(row => {
+          val result = new ArrayBuffer[String]()
+          val bucketsMap = bucketsMap_br.value
+          row.foreach{
+            case (label, features) =>
+              val featuresBucket = features.map{
+                case (name, score) =>
+                  var ifFilter = false
+                  if (filterNames.nonEmpty){
+                    filterNames.foreach(r=> if (!ifFilter && name.contains(r)) {ifFilter = true} )
+                  }
+                  if (ifFilter){
+                    ""
+                  }else{
+                    if (score > 1E-8) {
+                      if (bucketsMap.contains(name)) {
+                        val (bucketsNum, buckets) = bucketsMap(name)
+                        val scoreNew = 1.0 / bucketsNum * (ExtractorUtils.findInsertPosition(buckets, score).toDouble + 1.0)
+                        name + ":" + scoreNew.toString
+                      } else {
+                        var isModify = false
+                        if (modifyFeatureName.nonEmpty) {
+                          modifyFeatureName.foreach(r => if (!isModify && name.startsWith(r)) {
+                            isModify = true
+                          })
+                        }
+                        if (isModify) {
+                          name + ":" + defaultValue
+                        } else {
+                          name + ":" + score.toString
+                        }
+                      }
+                    } else {
+                      ""
+                    }
+                  }
+              }.filter(_.nonEmpty)
+              result.add(label + "\t" + featuresBucket.mkString("\t"))
+          }
+          result.iterator
+      })
+
+      // 4 保存数据到hdfs
+      val hdfsPath = savePath + "/" + date
+      if (hdfsPath.nonEmpty && hdfsPath.startsWith("/dw/recommend/model/")) {
+        println("删除路径并开始数据写入:" + hdfsPath)
+        MyHdfsUtils.delete_hdfs_path(hdfsPath)
+        data.repartition(repartition).saveAsTextFile(hdfsPath, classOf[GzipCodec])
+      } else {
+        println("路径不合法，无法写入:" + hdfsPath)
+      }
+    }
+
+
+
+  }
+}