行为预测性别

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

@@ -0,0 +1,121 @@
+package com.tzld.piaoquan.recommend.model
+
+import ml.dmlc.xgboost4j.scala.spark.XGBoostClassificationModel
+import org.apache.commons.lang.math.NumberUtils
+import org.apache.commons.lang3.StringUtils
+import org.apache.hadoop.io.compress.GzipCodec
+import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator
+import org.apache.spark.ml.feature.VectorAssembler
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.types.DataTypes
+import org.apache.spark.sql.{Row, SparkSession}
+
+import java.util
+import scala.io.Source
+
+
+object pred_profile_gender_xgb_20251114 {
+  def main(args: Array[String]): Unit = {
+    val spark = SparkSession
+      .builder()
+      .appName(this.getClass.getName)
+      .getOrCreate()
+    val sc = spark.sparkContext
+
+    val param = ParamUtils.parseArgs(args)
+    val modelPath = param.getOrElse("modelPath", "/dw/recommend/model/user_profile/gender/model/model_xgb")
+    val testPath = param.getOrElse("testPath", "")
+    val featureFile = param.getOrElse("featureFile", "20241209_recsys_rov_name.txt")
+    val minCnt = param.getOrElse("minCnt", "10").toDouble
+    val savePath = param.getOrElse("savePath", "/dw/recommend/model/user_profile/gender/result")
+    val repartition = param.getOrElse("repartition", "20").toInt
+
+    val features = loadFeatureNames(featureFile)
+    var fields = Array(
+      DataTypes.createStructField("label", DataTypes.IntegerType, true)
+    ) ++ features.map(f => DataTypes.createStructField(f, DataTypes.DoubleType, true))
+    fields = fields ++ Array(
+      DataTypes.createStructField("mid", DataTypes.StringType, true),
+      DataTypes.createStructField("cnt", DataTypes.IntegerType, true)
+    )
+    val schema = DataTypes.createStructType(fields)
+    val vectorAssembler = new VectorAssembler().setInputCols(features).setOutputCol("features")
+
+    val model = XGBoostClassificationModel.load(modelPath)
+    model.setMissing(0.0f).setFeaturesCol("features")
+    val testData = createData(
+      minCnt,
+      sc.textFile(testPath),
+      features
+    )
+
+    val testDataSet = spark.createDataFrame(testData, schema)
+    val testDataSetTrans = vectorAssembler.transform(testDataSet).select("features", "label", "mid", "cnt")
+    val predictions = model.transform(testDataSetTrans)
+    val saveData = predictions.select("label", "rawPrediction", "probability", "mid", "cnt").rdd
+      .map(r => {
+        (r.get(0), r.get(1), r.get(2), r.get(3), r.get(4)).productIterator.mkString("\t")
+      })
+    val hdfsPath = savePath
+    if (hdfsPath.nonEmpty && hdfsPath.startsWith("/dw/recommend/model/")) {
+      println("删除路径并开始数据写入:" + hdfsPath)
+      MyHdfsUtils.delete_hdfs_path(hdfsPath)
+      saveData.repartition(repartition).saveAsTextFile(hdfsPath, classOf[GzipCodec])
+    } else {
+      println("路径不合法，无法写入:" + hdfsPath)
+    }
+
+    val evaluator = new BinaryClassificationEvaluator()
+      .setLabelCol("label")
+      .setRawPredictionCol("probability")
+      .setMetricName("areaUnderROC")
+    val auc = evaluator.evaluate(predictions.select("label", "probability"))
+    println("recsys rov:auc:" + auc)
+
+    println("---------------------------------\n")
+    println("---------------------------------\n")
+  }
+
+  def createData(minCnt: Double, data: RDD[String], features: Array[String]): RDD[Row] = {
+    data
+      .map(row => {
+        val cells: Array[String] = StringUtils.split(row, '\t')
+        val mid = cells(0)
+        val label = NumberUtils.toInt(cells(1))
+        val featureMap: util.Map[String, Double] = new util.HashMap[String, Double]
+        for (i <- 2 until cells.length) {
+          val fv: Array[String] = StringUtils.split(cells(i), ':')
+          featureMap.put(fv(0), NumberUtils.toDouble(fv(1), 0.0))
+        }
+        (mid, label, featureMap)
+      })
+      .filter {
+        case (mid, label, featureMap) =>
+          val cnt = featureMap.getOrDefault("cnt", 0.0d)
+          cnt >= minCnt
+      }
+      .map {
+        case (mid, label, featureMap) =>
+          val v: Array[Any] = new Array[Any](features.length + 3)
+          v(0) = label
+          for (i <- features.indices) {
+            v(i + 1) = featureMap.getOrDefault(features(i), 0.0d)
+          }
+          v(features.length + 1) = mid
+          v(features.length + 2) = featureMap.getOrDefault("cnt", 0.0d).toInt
+          Row(v: _*)
+      }
+  }
+
+  def loadFeatureNames(nameFile: String): Array[String] = {
+    val buffer = Source.fromFile(nameFile)
+    val names = buffer.getLines().mkString("\n")
+    buffer.close()
+    val featArray = names.split("\n")
+      .map(r => r.replace(" ", "").replaceAll("\n", ""))
+      .filter(r => r.nonEmpty)
+    println("featArray.size=" + featArray.length)
+    println(featArray.mkString(","))
+    featArray
+  }
+}

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

@@ -0,0 +1,114 @@
+package com.tzld.piaoquan.recommend.model
+
+import ml.dmlc.xgboost4j.scala.spark.XGBoostClassifier
+import org.apache.commons.lang.math.NumberUtils
+import org.apache.commons.lang3.StringUtils
+import org.apache.spark.ml.feature.VectorAssembler
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.types.DataTypes
+import org.apache.spark.sql.{Dataset, Row, SparkSession}
+
+import java.util
+import scala.io.Source
+
+object train_profile_gender_xgb_20251114 {
+  def main(args: Array[String]): Unit = {
+    val spark = SparkSession
+      .builder()
+      .appName(this.getClass.getName)
+      .getOrCreate()
+    val sc = spark.sparkContext
+
+    val param = ParamUtils.parseArgs(args)
+    val trainPath = param.getOrElse("trainPath", "/dw/recommend/model/user_profile/gender/sample/train/2025_y")
+    val featureFile = param.getOrElse("featureFile", "20241209_recsys_nor_name.txt")
+    val minCnt = param.getOrElse("minCnt", "10").toDouble
+    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 modelPath = param.getOrElse("modelPath", "/dw/recommend/model/user_profile/gender/model/model_xgb")
+
+    val features = loadFeatureNames(featureFile)
+    val trainData = createData(
+      minCnt,
+      sc.textFile(trainPath),
+      features
+    )
+    println("profile gender:train data size：" + trainData.count())
+
+    val fields = Array(
+      DataTypes.createStructField("label", DataTypes.IntegerType, true)
+    ) ++ features.map(f => DataTypes.createStructField(f, DataTypes.DoubleType, true))
+
+    val schema = DataTypes.createStructType(fields)
+    val trainDataSet: Dataset[Row] = spark.createDataFrame(trainData, schema)
+    val vectorAssembler = new VectorAssembler().setInputCols(features).setOutputCol("features")
+    val xgbInput = vectorAssembler.transform(trainDataSet).select("features", "label").persist()
+    val xgbClassifier = new XGBoostClassifier()
+      .setEta(eta)
+      .setGamma(gamma)
+      .setMissing(0.0f)
+      .setMaxDepth(max_depth)
+      .setNumRound(num_round)
+      .setSubsample(0.8)
+      .setColsampleBytree(0.8)
+      .setScalePosWeight(1)
+      .setObjective(func_object)
+      .setEvalMetric(func_metric)
+      .setFeaturesCol("features")
+      .setLabelCol("label")
+      .setNthread(1)
+      .setNumWorkers(num_worker)
+      .setSeed(2024)
+      .setMinChildWeight(1)
+    val model = xgbClassifier.fit(xgbInput)
+    if (modelPath.nonEmpty) {
+      model.write.overwrite.save(modelPath)
+    }
+  }
+
+  def createData(minCnt: Double, data: RDD[String], features: Array[String]): RDD[Row] = {
+    data
+      .map(row => {
+        val cells: Array[String] = StringUtils.split(row, '\t')
+        val mid = cells(0)
+        val label = NumberUtils.toInt(cells(1))
+        val featureMap: util.Map[String, Double] = new util.HashMap[String, Double]
+        for (i <- 2 until cells.length) {
+          val fv: Array[String] = StringUtils.split(cells(i), ':')
+          featureMap.put(fv(0), NumberUtils.toDouble(fv(1), 0.0))
+        }
+        (mid, label, featureMap)
+      })
+      .filter {
+        case (mid, label, featureMap) =>
+          val cnt = featureMap.getOrDefault("cnt", 0.0d)
+          cnt >= minCnt
+      }
+      .map {
+        case (mid, label, featureMap) =>
+          val v: Array[Any] = new Array[Any](features.length + 1)
+          v(0) = label
+          for (i <- features.indices) {
+            v(i + 1) = featureMap.getOrDefault(features(i), 0.0d)
+          }
+          Row(v: _*)
+      }
+  }
+
+  def loadFeatureNames(nameFile: String): Array[String] = {
+    val buffer = Source.fromFile(nameFile)
+    val names = buffer.getLines().mkString("\n")
+    buffer.close()
+    val featArray = names.split("\n")
+      .map(r => r.replace(" ", "").replaceAll("\n", ""))
+      .filter(r => r.nonEmpty)
+    println("featArray.size=" + featArray.length)
+    println(featArray.mkString(","))
+    featArray
+  }
+}