feat:ros二分类任务分桶脚本

src/main/scala/com/aliyun/odps/spark/examples/makedata_recsys/v20250218/makedata_recsys_43_ros_binary_weight_data_bucket_20250304.scala

@@ -0,0 +1,134 @@
+package com.aliyun.odps.spark.examples.makedata_recsys.v20250218
+
+import com.alibaba.fastjson.JSON
+import com.aliyun.odps.spark.examples.myUtils.{FileUtils, 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
+
+/**
+ * ros模型退化为二分类问题，每个label>0的样本 扩展成n条二分类中的正样本与一条负样本
+ */
+object makedata_recsys_43_ros_binary_weight_data_bucket_20250304 {
+  def main(args: Array[String]): Unit = {
+
+    // 1 读取参数
+    val param = ParamUtils.parseArgs(args)
+    val readPath = param.getOrElse("readPath", "/dw/recommend/model/41_recsys_ros_train_data/")
+    val savePath = param.getOrElse("savePath", "/dw/recommend/model/43_recsys_ros_data_bucket_reg/")
+    val beginStr = param.getOrElse("beginStr", "20250224")
+    val endStr = param.getOrElse("endStr", "20250225")
+    val repartition = param.getOrElse("repartition", "100").toInt
+    val filterNames = param.getOrElse("filterNames", "").split(",").filter(_.nonEmpty).toSet
+    val noBucketFeature = param.getOrElse("noBucketFeature", "hour,is_greeting,day_of_week,return_n_uv").split(",").filter(_.nonEmpty).toSet
+    val whatLabel = param.getOrElse("whatLabel", "return_n_uv")
+    val whatApps = param.getOrElse("whatApps", "0,4,2,32,17,18,21,22,24,25,26,27,28,29,3,30,31,33,34,35,36").split(",").toSet
+    val fileName = param.getOrElse("fileName", "20250306_ros_bucket_229.txt")
+
+    val spark = SparkSession
+      .builder()
+      .appName(this.getClass.getName)
+      .getOrCreate()
+    val sc = spark.sparkContext
+
+    val loader = getClass.getClassLoader
+    val resourceUrlBucket = loader.getResource(fileName)
+    val buckets = FileUtils.readFile(resourceUrlBucket)
+    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)
+
+
+    val dateRange = MyDateUtils.getDateRange(beginStr, endStr)
+    for (date <- dateRange) {
+      println("开始执行:" + date)
+      println(readPath + "/" + 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 logJson = JSON.parseObject(logKey)
+            val appType = logJson.getString("apptype")
+            whatApps.contains(appType)
+        }
+        .map {
+          case (logKey, labelKey, features) =>
+            val labelJson = JSON.parseObject(labelKey)
+            val returnNUv = Integer.parseInt(labelJson.getOrDefault(whatLabel, "0").toString)
+
+            val logJson = JSON.parseObject(logKey)
+            logJson.put("return_n_uv", returnNUv)
+
+            val label = if (returnNUv > 0) 1 else 0
+            (logJson.toString(), label, features)
+
+        }
+        .mapPartitions(row => {
+          val result = new ArrayBuffer[String]()
+          val bucketsMap = bucketsMap_br.value
+          row.foreach {
+            case (logKey, 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 (noBucketFeature.nonEmpty && noBucketFeature.contains(name)) {
+                        name + ":" + score.toString
+                      } else {
+                        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 {
+                          ""
+                        }
+                      }
+                    } else {
+                      ""
+                    }
+                  }
+              }.filter(_.nonEmpty)
+              result.add(logKey + "\t" + 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)
+      }
+    }
+  }
+}