新增特征和处理方式

src/main/scala/com/aliyun/odps/spark/examples/makedata_recsys/makedata_recsys_41_str2ros_originData_20241209.scala

@@ -8,7 +8,7 @@ import examples.extractor.RankExtractorFeature_20240530
 import org.apache.hadoop.io.compress.GzipCodec
 import org.apache.spark.sql.SparkSession
 import org.xm.Similarity
-
+import examples.utils.SimilarityUtils
 import scala.collection.JavaConversions._
 import scala.collection.mutable.ArrayBuffer
 
@@ -26,15 +26,16 @@ object makedata_recsys_41_str2ros_originData_20241209 {
 
     // 1 读取参数
     val param = ParamUtils.parseArgs(args)
-    val tablePart = param.getOrElse("tablePart", "64").toInt
+
     val beginStr = param.getOrElse("beginStr", "2024120912")
     val endStr = param.getOrElse("endStr", "2024120912")
-    val savePath = param.getOrElse("savePath", "/dw/recommend/model/41_recsys_sample_str2ros_data_table")
     val project = param.getOrElse("project", "loghubods")
-    val table = param.getOrElse("table", "XXXX")
+    val table = param.getOrElse("table", "alg_recsys_sample_all_v2")
+    val tablePart = param.getOrElse("tablePart", "64").toInt
+    val savePath = param.getOrElse("savePath", "/dw/recommend/model/61_origin_data/")
     val repartition = param.getOrElse("repartition", "32").toInt
 
-    // 2 读取odps+表信息
+    // 2 odps
     val odpsOps = env.getODPS(sc)
 
     // 3 循环执行数据生产
@@ -127,7 +128,7 @@ object makedata_recsys_41_str2ros_originData_20241209 {
               for (key_time <- List("tags_1d", "tags_3d", "tags_7d")) {
                 val tags = if (c34567.containsKey(key_time)) c34567.getString(key_time) else ""
                 if (!tags.equals("")) {
-                  val (f1, f2, f3, f4) = funcC34567ForTags(tags, title)
+                  val (f1, f2, f3, f4) = funcC34567ForTagsW2V(tags, title)
                   featureMap.put(key_feature + "_" + key_time + "_matchnum", f1)
                   featureMap.put(key_feature + "_" + key_time + "_maxscore", f3)
                   featureMap.put(key_feature + "_" + key_time + "_avgscore", f4)
@@ -167,8 +168,8 @@ object makedata_recsys_41_str2ros_originData_20241209 {
           }
 
           // ************* new feature *************
-          val shortPeriod = List("1h", "2h", "3h", "4h", "6h", "12h", "24h", "7d")
-          val middlePeriod = List("1d", "7d", "14d", "30d")
+          val shortPeriod = List("1h", "2h", "4h", "6h", "12h", "24h")
+          val middlePeriod = List("7d", "14d", "30d")
           val longPeriod = List("7d", "35d", "90d", "365d")
           val vidStatFeat = List(
             ("b20", shortPeriod, getJsonObject(record, "b20_feature")),
@@ -183,7 +184,7 @@ object makedata_recsys_41_str2ros_originData_20241209 {
           )
           for ((featType, featPeriod, featData) <- vidStatFeat) {
             for (period <- featPeriod) {
-              val view = if (featData.isEmpty) 0D else featData.getDoubleValue("view_" + period)
+              // val view = if (featData.isEmpty) 0D else featData.getDoubleValue("view_" + period)
               val share = if (featData.isEmpty) 0D else featData.getDoubleValue("share_" + period)
               val return_ = if (featData.isEmpty) 0D else featData.getDoubleValue("return_" + period)
               val view_hasreturn = if (featData.isEmpty) 0D else featData.getDoubleValue("view_hasreturn_" + period)
@@ -195,13 +196,13 @@ object makedata_recsys_41_str2ros_originData_20241209 {
               val r_cnt4s = if (featData.isEmpty) 0D else featData.getDoubleValue("r_cnt4s_" + period)
               val str = if (featData.isEmpty) 0D else featData.getDoubleValue("str_" + period)
               // scale
-              val view_s = RankExtractorFeature_20240530.calLog(view)
+              // val view_s = RankExtractorFeature_20240530.calLog(view)
               val share_s = RankExtractorFeature_20240530.calLog(share)
               val return_s = RankExtractorFeature_20240530.calLog(return_)
               val view_hasreturn_s = RankExtractorFeature_20240530.calLog(view_hasreturn)
               val share_hasreturn_s = RankExtractorFeature_20240530.calLog(share_hasreturn)
 
-              featureMap.put(featType + "_" + period + "_" + "view", view_s)
+              // featureMap.put(featType + "_" + period + "_" + "view", view_s)
               featureMap.put(featType + "_" + period + "_" + "share", share_s)
               featureMap.put(featType + "_" + period + "_" + "return", return_s)
               featureMap.put(featType + "_" + period + "_" + "view_hasreturn", view_hasreturn_s)
@@ -245,31 +246,27 @@ object makedata_recsys_41_str2ros_originData_20241209 {
           }
 
           /*
+          视频特征： 5*6*5 = 240个
+                    曝光使用pv 分享使用pv 回流使用uv --> 1h 2h 3h 4h 12h 1d 3d 7d
+                    STR log(share) ROV log(return) ROV*log(return) ROS
+                    整体、整体曝光对应、推荐非冷启root、推荐冷启root、分省份root
+          视频基础： 2个   视频时长、比特率
+          用户： 4+8 = 12个
+                    播放次数 --> 6h 1d 3d 7d --> 4个
+                    带回来的分享pv 回流uv --> 12h 1d 3d 7d --> 8个
+          人+vid-title:  5*3*3 = 45
+                    播放点/回流点/分享点/累积分享/累积回流 --> 1d 3d 7d --> 匹配数量 语义最高相似度分 语义平均相似度分 --> 45个
+          人+vid-cf： 2*3*3 = 12
+                    基于分享行为/基于回流行为 -->  “分享cf”+”回流点击cf“ 相似分 相似数量 相似rank的倒数 --> 12个
+          头部视频：  3
+                    曝光 回流 ROVn 3个特征
+          场景：     小时 星期 apptype city province pagesource 机器型号
+          总量：     240+2+12+45+12+3 = 314
+          ---------------------------------------------------------------
+          视频特征：(6+3+4)*10 = 130个
+          CF: 13个
 
 
-          视频：
-          曝光使用pv 分享使用pv 回流使用uv --> 1h 2h 3h 4h 12h 1d 3d 7d
-          STR log(share) ROV log(return) ROV*log(return)
-          40个特征组合
-          整体、整体曝光对应、推荐非冷启root、推荐冷启root、分省份root
-          200个特征值
-
-          视频：
-          视频时长、比特率
-
-          人：
-          播放次数 --> 6h 1d 3d 7d --> 4个
-          带回来的分享pv 回流uv --> 12h 1d 3d 7d --> 8个
-          人+vid-title:
-          播放点/回流点/分享点/累积分享/累积回流 --> 1d 3d 7d --> 匹配数量 语义最高相似度分 语义平均相似度分 --> 45个
-          人+vid-cf
-          基于分享行为/基于回流行为 -->  “分享cf”+”回流点击cf“ 相似分 相似数量 相似rank的倒数 --> 12个
-
-          头部视频：
-          曝光 回流 ROVn 3个特征
-
-          场景：
-          小时 星期 apptype city province pagesource 机器型号
            */
 
 
@@ -277,7 +274,7 @@ object makedata_recsys_41_str2ros_originData_20241209 {
           val labels = new JSONObject
           for (labelKey <- List(
             "is_play", "is_share", "is_return", "noself_is_return", "return_uv", "noself_return_uv", "total_return_uv",
-            "share_pv", "total_share_uv"
+            "share_pv", "total_share_uv", "view_24h", "total_return_uv_new"
           )) {
             if (!record.isNull(labelKey)) {
               labels.put(labelKey, record.getString(labelKey))
@@ -317,8 +314,14 @@ object makedata_recsys_41_str2ros_originData_20241209 {
   }
 
   def getJsonObject(record: Record, key: String): JSONObject = {
-    if (record.isNull(key)) new JSONObject() else
-      JSON.parseObject(record.getString(key))
+    val data = if (record.isNull(key)) new JSONObject() else JSON.parseObject(record.getString(key))
+    val data2 = new JSONObject()
+    data.foreach(r => {
+      if (r._2 != null){
+        data2.put(r._1, r._2)
+      }
+    })
+    data2
   }
 
   def getJsonObject(obj: JSONObject, keyName: String, valueName: String): JSONObject = {
@@ -336,10 +339,10 @@ object makedata_recsys_41_str2ros_originData_20241209 {
         }
       }
     }
-    return map
+    map
   }
 
-  def funcC34567ForTags(tags: String, title: String): Tuple4[Double, String, Double, Double] = {
+  def funcC34567ForTagsW2V(tags: String, title: String): Tuple4[Double, String, Double, Double] = {
     // 匹配数量 匹配词 语义最高相似度分 语义平均相似度分
     val tagsList = tags.split(",")
     var d1 = 0.0
@@ -351,7 +354,7 @@ object makedata_recsys_41_str2ros_originData_20241209 {
         d1 = d1 + 1.0
         d2.add(tag)
       }
-      val score = Similarity.conceptSimilarity(tag, title)
+      val score = SimilarityUtils.word2VecSimilarity(tag, title)
       d3 = if (score > d3) score else d3
       d4 = d4 + score
     }