scala train

recommend-model-produce/src/main/scala/com/tzld/piaoquan/recommend/model/train_01_xgb_ad_20240808.scala

@@ -10,7 +10,9 @@ import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.types.{DataTypes, StructField}
 import org.apache.spark.sql.{Dataset, Row, RowFactory, SparkSession}
 
+import scala.collection.JavaConversions._
 import java.util
+import scala.collection.mutable.ArrayBuffer
 import scala.io.Source
 
 object train_01_xgb_ad_20240808{
@@ -20,10 +22,24 @@ object train_01_xgb_ad_20240808{
       .appName(this.getClass.getName)
       .getOrCreate()
     val sc = spark.sparkContext
-//    val features = Array("cpa", "b2_12h_ctr", "b2_12h_ctcvr", "b2_12h_cvr", "b2_12h_conver", "b2_12h_click", "b2_12h_conver*log(view)", "b2_12h_conver*ctcvr", "b2_7d_ctr", "b2_7d_ctcvr", "b2_7d_cvr", "b2_7d_conver", "b2_7d_click", "b2_7d_conver*log(view)", "b2_7d_conver*ctcvr")
+
+    val param = ParamUtils.parseArgs(args)
+    val featureFile = param.getOrElse("featureFile", "20240703_ad_feature_name.txt")
+    val trainPath = param.getOrElse("trainPath", "/dw/recommend/model/33_ad_train_data_v4/20240724")
+    val testPath = param.getOrElse("testPath", "/dw/recommend/model/33_ad_train_data_v4/20240725")
+    val savePath = param.getOrElse("savePath", "/dw/recommend/model/34_ad_predict_data/")
+    val featureFilter = param.getOrElse("featureFilter", "XXXXXX").split(",")
+    val eta = param.getOrElse("eta", "0.01").toDouble
+    val gamma = param.getOrElse("gamma", "0.0").toDouble
+    val max_depth = param.getOrElse("max_depth", "5").toInt
+    val num_round = param.getOrElse("num_round", "100").toInt
+    val num_worker = param.getOrElse("num_worker", "20").toInt
+    val func_object = param.getOrElse("func_object", "binary:logistic")
+    val func_metric = param.getOrElse("func_metric", "auc")
+    val repartition = param.getOrElse("repartition", "20").toInt
 
     val loader = getClass.getClassLoader
-    val resourceUrl = loader.getResource("20240703_ad_feature_name.txt")
+    val resourceUrl = loader.getResource(featureFile)
     val content =
       if (resourceUrl != null) {
         val content = Source.fromURL(resourceUrl).getLines().mkString("\n")
@@ -33,18 +49,20 @@ object train_01_xgb_ad_20240808{
         ""
       }
     println(content)
+
     val features = content.split("\n")
       .map(r => r.replace(" ", "").replaceAll("\n", ""))
-      .filter(r => r.nonEmpty)
-
+      .filter(r => r.nonEmpty || !featureFilter.contains(r))
+    println("features.size=" + features.length)
 
-    val trainData = createData(
-      sc.textFile("/dw/recommend/model/33_ad_train_data_v4/20240724"),
+    val trainData = createData4Ad(
+      sc.textFile(trainPath),
       features
     )
-    println("train data size：" + trainData.count())
+    println("zhangbo:train data size：" + trainData.count())
 
     val fields = Array(
+      DataTypes.createStructField("logKey", DataTypes.StringType, true),
       DataTypes.createStructField("label", DataTypes.IntegerType, true)
     ) ++ features.map(f => DataTypes.createStructField(f, DataTypes.DoubleType, true))
     val schema = DataTypes.createStructType(fields)
@@ -56,39 +74,40 @@ object train_01_xgb_ad_20240808{
 //      "objective" -> "binary:logistic",
 //      "num_class" -> 3)
     val xgbClassifier = new XGBoostClassifier()
-      .setEta(0.01f)
+      .setEta(eta)
+      .setGamma(gamma)
       .setMissing(0.0f)
-      .setMaxDepth(5)
-      .setNumRound(1000)
+      .setMaxDepth(max_depth)
+      .setNumRound(num_round)
       .setSubsample(0.8)
       .setColsampleBytree(0.8)
-      .setScalePosWeight(1)
-      .setObjective("binary:logistic")
-      .setEvalMetric("auc")
+//      .setScalePosWeight(1)
+      .setObjective(func_object)
+      .setEvalMetric(func_metric)
       .setFeaturesCol("features")
       .setLabelCol("label")
       .setNthread(1)
-      .setNumWorkers(22)
+      .setNumWorkers(num_worker)
     val model = xgbClassifier.fit(xgbInput)
 
 
-    val testData = createData(
-      sc.textFile("/dw/recommend/model/33_ad_train_data_v4/20240725"),
+    val testData = createData4Ad(
+      sc.textFile(testPath),
       features
     )
     val testDataSet = spark.createDataFrame(testData, schema)
     val testDataSetTrans = vectorAssembler.transform(testDataSet).select("features","label")
     val predictions = model.transform(testDataSetTrans)
 
-    val saveData = predictions.select("label", "prediction", "features", "rawPrediction", "probability").rdd
+    val saveData = predictions.select("label", "logKey", "rawPrediction", "probability").rdd
       .map(r =>{
-        (r.get(0), r.get(1), r.get(2), r.get(3), r.get(4)).productIterator.mkString("\t")
+        (r.get(0), r.get(1), r.get(2), r.get(3)).productIterator.mkString("\t")
     })
-    val hdfsPath = "/dw/recommend/model/checkpoint_xgbtest"
+    val hdfsPath = savePath
     if (hdfsPath.nonEmpty && hdfsPath.startsWith("/dw/recommend/model/")) {
       println("删除路径并开始数据写入:" + hdfsPath)
       MyHdfsUtils.delete_hdfs_path(hdfsPath)
-      saveData.repartition(100).saveAsTextFile(hdfsPath, classOf[GzipCodec])
+      saveData.repartition(repartition).saveAsTextFile(hdfsPath, classOf[GzipCodec])
     } else {
       println("路径不合法，无法写入:" + hdfsPath)
     }
@@ -104,22 +123,27 @@ object train_01_xgb_ad_20240808{
   }
 
 
-  def createData(data: RDD[String], features: Array[String]): RDD[Row] = {
+  def createData4Ad(data: RDD[String], features: Array[String]): RDD[Row] = {
     data.map(r => {
-      val line: Array[String] = StringUtils.split(r, '\t')
-      val label: Int = NumberUtils.toInt(line(0))
-      val map: util.Map[String, Double] = new util.HashMap[String, Double]
-      for (i <- 1 until line.length) {
-        val fv: Array[String] = StringUtils.split(line(i), ':')
-        map.put(fv(0), NumberUtils.toDouble(fv(1), 0.0))
-      }
-
-      val v: Array[Any] = new Array[Any](features.length + 1)
-      v(0) = label
-      for (i <- 0 until features.length) {
-        v(i + 1) = map.getOrDefault(features(i), 0.0d)
-      }
-      Row(v: _*)
+      val rList = r.split("\t")
+      val label = rList(0).toInt
+      val featureMap = scala.collection.mutable.Map[String, Double]()
+      var cid = "-1"
+      rList.drop(1).foreach(kv =>{
+        val kv_ = kv.split(":")
+        if (kv_(0).startsWith("cid_")){
+          cid = kv_(0).split("_")(1)
+        }else{
+          featureMap.put(kv_(0), kv_(1).toDouble)
+        }
+      })
+      val res = new ArrayBuffer[Any]()
+      res.add(cid)
+      res.add(label)
+      features.foreach(r =>{
+        res.add(featureMap.getOrElse(r, 0.0D))
+      })
+      Row(res)
     })
   }
 }