add 71 rov sample

src/main/scala/com/aliyun/odps/spark/examples/makedata_recsys_r_rate/makedata_recsys_71_originData_20250109.scala

@@ -27,16 +27,15 @@ object makedata_recsys_71_originData_20250109 {
 
     // 1 读取参数
     val param = ParamUtils.parseArgs(args)
-
     val beginStr = param.getOrElse("beginStr", "2025010723")
     val endStr = param.getOrElse("endStr", "2025010723")
     val project = param.getOrElse("project", "loghubods")
     val table = param.getOrElse("table", "alg_recsys_sample_tmp_20250109")
     val tablePart = param.getOrElse("tablePart", "64").toInt
-    val savePath = param.getOrElse("savePath", "/dw/recommend/model/71_origin_data/")
     val repartition = param.getOrElse("repartition", "32").toInt
     val whatLabel = param.getOrElse("whatLabel", "is_return_1")
     val fuSampleRate = param.getOrElse("fuSampleRate", "0.1").toDouble
+    val savePath = param.getOrElse("savePath", "/dw/recommend/model/71_origin_data/")
 
     // 2 odps
     val odpsOps = env.getODPS(sc)
@@ -54,14 +53,13 @@ object makedata_recsys_71_originData_20250109 {
           transfer = func,
           numPartition = tablePart)
         .filter(record => {
-          val label = if (record.isNull(whatLabel)) "0" else record.getString(whatLabel)
+          val label = getStringValue(record, whatLabel, "0")
           "1".equals(label) || new Random().nextDouble() <= fuSampleRate
         })
         .mapPartitions(p => {
           SimilarityUtils.init()
           p.map(record => {
             val featureMap = new JSONObject()
-
             val metaData = getJsonObject(record, "metafeaturemap")
             // a 视频特征
             val b1: JSONObject = getJsonObject(metaData, "alg_vid_feature_all_exp_v2")
@@ -311,15 +309,16 @@ object makedata_recsys_71_originData_20250109 {
             }
             //5 处理log key表头。
             val apptype = record.getString("apptype")
-            val pagesource = record.getString("pagesource")
+            val page = getStringValue(record, "page")
+            val pagesource = getStringValue(record, "pagesource")
+            val recommendpagetype = getStringValue(record, "recommendpagetype")
+            val flowpool = getStringValue(record, "flowpool")
+            val abcode = record.getString("abcode")
             val mid = record.getString("mid")
-            // vid 已经提取了
+            val level = getStringValue(record, "level", "0")
             val ts = record.getString("ts")
-            val abcode = record.getString("abcode")
-            val level = if (record.isNull("level")) "0" else record.getString("level")
-            val logKey = (apptype, pagesource, mid, vid, ts, abcode, level).productIterator.mkString(",")
+            val logKey = (apptype, page, pagesource, recommendpagetype, flowpool, abcode, mid, vid, level, ts).productIterator.mkString(",")
             val labelKey = labels.toString()
-            // val featureKey = featureMap.toString()
             val featureKey = truncateDecimal(featureMap).toString()
             val scoresMap = getSubJson(record, "extend_alg", "scoresMap").toString()
             //6 拼接数据，保存。
@@ -407,7 +406,6 @@ object makedata_recsys_71_originData_20250109 {
     data
   }
 
-
   def funcC34567ForTagsW2V(tags: String, title: String): Tuple4[Double, String, Double, Double] = {
     // 匹配数量 匹配词 语义最高相似度分 语义平均相似度分
     val tagsList = tags.split(",")
@@ -438,4 +436,12 @@ object makedata_recsys_71_originData_20250109 {
     }
     new JSONObject()
   }
+
+  def getStringValue(record: Record, key: String, default: String = ""): String = {
+    if (!record.isNull(key)) {
+      val value = record.getString(key)
+      return value.trim
+    }
+    default
+  }
 }

src/main/scala/com/aliyun/odps/spark/examples/makedata_recsys_r_rate/makedata_recsys_71_rov_sample_20250109.scala

@@ -0,0 +1,193 @@
+package com.aliyun.odps.spark.examples.makedata_recsys_r_rate
+
+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
+import scala.util.Random
+
+
+object makedata_recsys_71_rov_sample_20250109 {
+  def main(args: Array[String]): Unit = {
+
+    // 1 读取参数
+    val param = ParamUtils.parseArgs(args)
+    val readPath = param.getOrElse("readPath", "/dw/recommend/model/71_origin_data/")
+    val savePath = param.getOrElse("savePath", "/dw/recommend/model/71_recsys_rov_train_data/")
+    val beginStr = param.getOrElse("beginStr", "20250103")
+    val endStr = param.getOrElse("endStr", "20250103")
+    val repartition = param.getOrElse("repartition", "100").toInt
+    val whatLabel = param.getOrElse("whatLabel", "is_return_1")
+    val whatApps = param.getOrElse("whatApps", "0,3,4,21,17").split(",").toSet
+    val fuSampleRate = param.getOrElse("fuSampleRate", "1.0").toDouble
+    val notUseBucket = param.getOrElse("notUseBucket", "0").toInt
+    val featureNameFile = param.getOrElse("featureName", "20241209_recsys_rov_name.txt")
+    val featureBucketFile = param.getOrElse("featureBucket", "20241209_recsys_rov_bucket.txt")
+
+    val spark = SparkSession
+      .builder()
+      .appName(this.getClass.getName)
+      .getOrCreate()
+    val sc = spark.sparkContext
+
+    val loader = getClass.getClassLoader
+    val featureNameSet = loadUseFeatureNames(loader, featureNameFile)
+    val featureBucketMap = loadUseFeatureBuckets(loader, notUseBucket, featureBucketFile)
+    val bucketsMap_br = sc.broadcast(featureBucketMap)
+
+    val dateRange = MyDateUtils.getDateRange(beginStr, endStr)
+    for (date <- dateRange) {
+      println("开始执行:" + date)
+      val data = sc.textFile(readPath + "/" + date + "*").map(r => {
+          // logKey + "\t" + labelKey + "\t" + scoresMap + "\t" + featureKey
+          val rList = r.split("\t")
+          val logKey = rList(0)
+          val labelKey = rList(1)
+          val scoresMap = rList(2)
+          val featData = rList(3)
+          (logKey, labelKey, scoresMap, featData)
+        })
+        .filter {
+          case (logKey, labelKey, scoresMap, featData) =>
+            validData(logKey, whatApps)
+        }.filter {
+          case (logKey, labelKey, scoresMap, featData) =>
+            val label = parseLabel(labelKey, whatLabel)
+            "1".equals(label) || new Random().nextDouble() <= fuSampleRate
+        }
+        .map {
+          case (logKey, labelKey, scoresMap, featData) =>
+            val label = parseLabel(labelKey, whatLabel)
+            val features = parseFeature(featData)
+            (logKey, label, scoresMap, features)
+        }
+        .mapPartitions(row => {
+          val result = new ArrayBuffer[String]()
+          val bucketsMap = bucketsMap_br.value
+          row.foreach {
+            case (logKey, label, scoresMap, features) =>
+              val featuresBucket = bucketFeature(featureNameSet, bucketsMap, features)
+              result.add(logKey + "\t" + label + "\t" + scoresMap + "\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)
+      }
+    }
+  }
+
+  private def loadFileData(loader: ClassLoader, nameFile: String): String = {
+    val resourceUrlBucket = loader.getResource(nameFile)
+    val data =
+      if (resourceUrlBucket != null) {
+        val buckets = Source.fromURL(resourceUrlBucket).getLines().mkString("\n")
+        Source.fromURL(resourceUrlBucket).close()
+        buckets
+      } else {
+        ""
+      }
+    data
+  }
+
+  private def loadUseFeatureNames(loader: ClassLoader, nameFile: String): Set[String] = {
+    val names = loadFileData(loader, nameFile)
+    println(names)
+    names.split("\n")
+      .map(r => r.replace(" ", "").replaceAll("\n", ""))
+      .filter(r => r.nonEmpty)
+      .toSet
+  }
+
+  private def loadUseFeatureBuckets(loader: ClassLoader, notUseBucket: Int, nameFile: String): Map[String, (Double, Array[Double])] = {
+    if (notUseBucket > 0) {
+      return scala.collection.mutable.Map[String, (Double, Array[Double])]
+    }
+    val buckets = loadFileData(loader, nameFile)
+    println(buckets)
+    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
+  }
+
+  private def recommendFlow(flowPool: String): Boolean = {
+    if (flowPool.isEmpty || flowPool.endsWith("#1")) {
+      return true
+    }
+    false
+  }
+
+  private def validData(logKey: String, whatApps: Set[String]): Boolean = {
+    // apptype, page, pagesource, recommendpagetype, flowpool, abcode, mid, vid, level, ts
+    val cells = logKey.split(",")
+    val apptype = cells(0)
+    val page = cells(1)
+    //val pagesource = cells(2)
+    val recommendpagetype = cells(3)
+    val flowpool = cells(4)
+    if (whatApps.contains(apptype)) {
+      if (recommendFlow(flowpool)) {
+        if (page.equals("详情后沉浸页")) {
+          return true
+        } else if (page.equals("回流后沉浸页&内页feed")) {
+          if (recommendpagetype.endsWith("recommend-detail")) {
+            return true
+          }
+        }
+      }
+    }
+    false
+  }
+
+  private def parseLabel(data: String, key: String, default: String = "0"): String = {
+    JSON.parseObject(data).getOrDefault(key, default).toString
+  }
+
+  private def parseFeature(data: String): scala.collection.mutable.Map[String, Double] = {
+    val features = scala.collection.mutable.Map[String, Double]()
+    if (data.nonEmpty) {
+      val obj = JSON.parseObject(data)
+      obj.foreach(r => {
+        features.put(r._1, obj.getDoubleValue(r._1))
+      })
+    }
+    features
+  }
+
+  private def bucketFeature(nameSet: Set[String], bucketMap: Map[String, (Double, Array[Double])], features: scala.collection.mutable.Map[String, Double]): Iterable[String] = {
+    features.map {
+      case (name, score) =>
+        if (!nameSet.contains(name)) {
+          ""
+        } else {
+          if (score > 1E-8) {
+            if (bucketMap.contains(name)) {
+              val (bucketsNum, buckets) = bucketMap(name)
+              val scoreNew = 1.0 / bucketsNum * (ExtractorUtils.findInsertPosition(buckets, score).toDouble + 1.0)
+              name + ":" + scoreNew.toString
+            } else {
+              name + ":" + score.toString
+            }
+          } else {
+            ""
+          }
+        }
+    }.filter(_.nonEmpty)
+  }
+}