nor xgb model

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

@@ -0,0 +1,112 @@
+package com.tzld.piaoquan.recommend.model
+
+import ml.dmlc.xgboost4j.scala.spark.XGBoostRegressionModel
+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.RegressionEvaluator
+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_recsys_61_xgb_nor_hdfsfile_20241209 {
+  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 featureFile = param.getOrElse("featureFile", "20241209_recsys_nor_name.txt")
+    val testPath = param.getOrElse("testPath", "")
+    val savePath = param.getOrElse("savePath", "/dw/recommend/model/61_recsys_nor_predict_data/")
+    val featureFilter = param.getOrElse("featureFilter", "XXXXXX").split(",")
+
+    val repartition = param.getOrElse("repartition", "20").toInt
+    val modelPath = param.getOrElse("modelPath", "/dw/recommend/model/61_recsys_nor_model/model_xgb")
+
+    val loader = getClass.getClassLoader
+    val resourceUrl = loader.getResource(featureFile)
+    val content =
+      if (resourceUrl != null) {
+        val content = Source.fromURL(resourceUrl).getLines().mkString("\n")
+        Source.fromURL(resourceUrl).close()
+        content
+      } else {
+        ""
+      }
+    println(content)
+
+    val features = content.split("\n")
+      .map(r => r.replace(" ", "").replaceAll("\n", ""))
+      .filter(r => r.nonEmpty || !featureFilter.contains(r))
+    println("features.size=" + features.length)
+
+    val fields = Array(
+      DataTypes.createStructField("label", DataTypes.IntegerType, true)
+    ) ++ features.map(f => DataTypes.createStructField(f, DataTypes.DoubleType, true))
+
+    val schema = DataTypes.createStructType(fields)
+    val vectorAssembler = new VectorAssembler().setInputCols(features).setOutputCol("features")
+
+    val model = XGBoostRegressionModel.load(modelPath)
+    model.setMissing(0.0f).setFeaturesCol("features")
+
+    val testData = createData(
+      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", "originalPrediction").rdd
+      .map(r => {
+        (r.get(0), r.get(1)).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 RegressionEvaluator()
+      .setLabelCol("label")
+      .setPredictionCol("originalPrediction")
+      .setMetricName("rmse")
+    val rmse = evaluator.evaluate(predictions.select("label", "originalPrediction"))
+    println("recsys rov:rmse:" + rmse)
+
+    println("---------------------------------\n")
+    println("---------------------------------\n")
+  }
+
+  def createData(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: _*)
+    })
+  }
+
+}

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

@@ -72,7 +72,7 @@ object train_recsys_61_xgb_nor_20241209 {
     //    val xgbParam = Map("eta" -> 0.01f,
     //      "max_depth" -> 5,
     //      "objective" -> "reg:squaredlogerror")
-    val xgbClassifier = new XGBoostRegressor()
+    val xgbRegressor = new XGBoostRegressor()
       .setEta(eta)
       .setGamma(gamma)
       .setMissing(0.0f)
@@ -88,7 +88,7 @@ object train_recsys_61_xgb_nor_20241209 {
       .setNumWorkers(num_worker)
       .setSeed(2024)
       .setMinChildWeight(1)
-    val model = xgbClassifier.fit(xgbInput)
+    val model = xgbRegressor.fit(xgbInput)
 
     if (modelPath.nonEmpty && modelFile.nonEmpty) {
       model.write.overwrite.save(modelPath)
@@ -108,7 +108,7 @@ object train_recsys_61_xgb_nor_20241209 {
       println("recsys nor:columns:" + predictions.columns.mkString(",")) //[label, features, originalPrediction]
       val saveData = predictions.select("label", "originalPrediction").rdd
         .map(r => {
-          (r.get(0), r.get(1), r.get(2)).productIterator.mkString("\t")
+          (r.get(0), r.get(1)).productIterator.mkString("\t")
         })
       val hdfsPath = savePath
       if (hdfsPath.nonEmpty && hdfsPath.startsWith("/dw/recommend/model/")) {