feat:新推荐特征处理

src/main/java/examples/extractor/ExtractorUtils.java

@@ -170,6 +170,13 @@ public class ExtractorUtils {
         return result;
     }
 
+    public static double reciprocal(double num) {
+        if (num == 0) {
+            throw new IllegalArgumentException("不能对零取倒数");
+        }
+        return 1.0 / num;
+    }
+
     public static void main(String[] args) {
         double[] sortedArray = {1.0, 2.0, 4.0, 4.0, 6.0};
         double target = 0.0;

src/main/java/examples/extractor/v20250218/ExtractItemFeature.java → src/main/java/examples/extractor/v20250218/ExtractFeature20250218.java

@@ -10,9 +10,9 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-public class ExtractItemFeature {
+public class ExtractFeature20250218 {
 
-    private ExtractItemFeature() {
+    private ExtractFeature20250218() {
     }
 
     public static void handleB1ToB13(Map<String, Map<String, Object>> videoFeature, Map<String, Object> featureMap) {
@@ -167,6 +167,24 @@ public class ExtractItemFeature {
         return resultMap;
     }
 
+    public static void useC6ToC7(Map<String, Map<String, String[]>> c67Map, String vid, Map<String, Object> featureMap) {
+        if (StringUtils.isBlank(vid)) {
+            return;
+        }
+        for (String key : Arrays.asList("c6", "c7")) {
+            for (String action : Arrays.asList("share", "return")) {
+                String featureKey = key + "_" + action;
+                if (c67Map.containsKey(featureKey)) {
+                    Map<String, String[]> cfMap = c67Map.get(featureKey);
+                    String[] scores = cfMap.get(vid);
+                    featureMap.put(featureKey + "_score", Double.parseDouble(scores[0]));
+                    featureMap.put(featureKey + "_num", Double.parseDouble(scores[1]));
+                    featureMap.put(featureKey + "_rank", ExtractorUtils.reciprocal(Double.parseDouble(scores[2])));
+                }
+            }
+        }
+    }
+
     public static void handleD3(Map<String, Object> d3Feature, Map<String, Object> featureMap) {
         for (String index : Arrays.asList("exp", "return_n", "rovn")) {
             double value = Double.parseDouble(d3Feature.getOrDefault(index, "0").toString());
@@ -206,13 +224,13 @@ public class ExtractItemFeature {
         String videoMergeCate1 = videoFeature.getOrDefault("merge_first_level_cate", "").toString();
         String videoFestiveLabel2 = videoFeature.getOrDefault("festive_label2", "").toString();
 
-        double titleSimilarity = ExtractItemFeature.calcTxtSimilarity(headVideoTitle, videoTitle);
-        double headTitleAndMerge1Similarity = ExtractItemFeature.calcTxtSimilarity(headVideoTitle, videoMergeCate1);
-        double headTitleAndMerge2Similarity = ExtractItemFeature.calcTxtSimilarity(headVideoTitle, videoMergeCate2);
-        double headTitleAndFestiveSimilarity = ExtractItemFeature.calcTxtSimilarity(headVideoTitle, videoFestiveLabel2);
-        double merge1Similarity = ExtractItemFeature.calcTxtSimilarity(headVideoMergeCate1, videoMergeCate1);
-        double merge2Similarity = ExtractItemFeature.calcTxtSimilarity(headVideoMergeCate2, videoMergeCate2);
-        double festiveSimilarity = ExtractItemFeature.calcTxtSimilarity(headVideoFestiveLabel2, videoFestiveLabel2);
+        double titleSimilarity = ExtractFeature20250218.calcTxtSimilarity(headVideoTitle, videoTitle);
+        double headTitleAndMerge1Similarity = ExtractFeature20250218.calcTxtSimilarity(headVideoTitle, videoMergeCate1);
+        double headTitleAndMerge2Similarity = ExtractFeature20250218.calcTxtSimilarity(headVideoTitle, videoMergeCate2);
+        double headTitleAndFestiveSimilarity = ExtractFeature20250218.calcTxtSimilarity(headVideoTitle, videoFestiveLabel2);
+        double merge1Similarity = ExtractFeature20250218.calcTxtSimilarity(headVideoMergeCate1, videoMergeCate1);
+        double merge2Similarity = ExtractFeature20250218.calcTxtSimilarity(headVideoMergeCate2, videoMergeCate2);
+        double festiveSimilarity = ExtractFeature20250218.calcTxtSimilarity(headVideoFestiveLabel2, videoFestiveLabel2);
 
         featureMap.put("title_sim", titleSimilarity);
         featureMap.put("head_title_merge1_sim", headTitleAndMerge1Similarity);

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

@@ -4,13 +4,15 @@ import com.alibaba.fastjson.{JSON, JSONObject}
 import com.aliyun.odps.TableSchema
 import com.aliyun.odps.data.Record
 import com.aliyun.odps.spark.examples.myUtils.{MyDateUtils, MyHdfsUtils, ParamUtils, env}
-import examples.extractor.RankExtractorFeature_20240530
+import examples.extractor.v20250218.ExtractFeature20250218
 import org.apache.hadoop.io.compress.GzipCodec
 import org.apache.spark.sql.SparkSession
 import org.xm.Similarity
 
+import java.util
 import scala.collection.JavaConversions._
 import scala.collection.mutable.ArrayBuffer
+
 /*
    20250218 提取特征
  */
@@ -30,7 +32,7 @@ object makedata_recsys_41_originData_20250218 {
     val endStr = param.getOrElse("endStr", "2025021812")
     val project = param.getOrElse("project", "loghubods")
     val table = param.getOrElse("table", "dwd_recsys_alg_sample_all_20250212")
-    val fuSampleRate = param.getOrElse("fuSampleRate", "0.1").toDouble  // 采样率1/20=0.05 自己不是唯一回流
+    val fuSampleRate = param.getOrElse("fuSampleRate", "0.1").toDouble // 采样率1/20=0.05 自己不是唯一回流
     val repartition = param.getOrElse("repartition", "32").toInt
     val savePath = param.getOrElse("savePath", "/dw/recommend/model/41_sample_data/")
 
@@ -52,169 +54,71 @@ object makedata_recsys_41_originData_20250218 {
         .map(record => {
 
           val featureMap = new JSONObject()
-
-          // a 视频特征
-          val b1: JSONObject = if (record.isNull("b1_feature")) new JSONObject() else
-            JSON.parseObject(record.getString("b1_feature"))
-          val b2: JSONObject = if (record.isNull("b2_feature")) new JSONObject() else
-            JSON.parseObject(record.getString("b2_feature"))
-          val b3: JSONObject = if (record.isNull("b3_feature")) new JSONObject() else
-            JSON.parseObject(record.getString("b3_feature"))
-          val b6: JSONObject = if (record.isNull("b6_feature")) new JSONObject() else
-            JSON.parseObject(record.getString("b6_feature"))
-          val b7: JSONObject = if (record.isNull("b7_feature")) new JSONObject() else
-            JSON.parseObject(record.getString("b7_feature"))
-
-          val b8: JSONObject = if (record.isNull("b8_feature")) new JSONObject() else
-            JSON.parseObject(record.getString("b8_feature"))
-          val b9: JSONObject = if (record.isNull("b9_feature")) new JSONObject() else
-            JSON.parseObject(record.getString("b9_feature"))
-          val b10: JSONObject = if (record.isNull("b10_feature")) new JSONObject() else
-            JSON.parseObject(record.getString("b10_feature"))
-          val b11: JSONObject = if (record.isNull("b11_feature")) new JSONObject() else
-            JSON.parseObject(record.getString("b11_feature"))
-          val b12: JSONObject = if (record.isNull("b12_feature")) new JSONObject() else
-            JSON.parseObject(record.getString("b12_feature"))
-          val b13: JSONObject = if (record.isNull("b13_feature")) new JSONObject() else
-            JSON.parseObject(record.getString("b13_feature"))
-          val b17: JSONObject = if (record.isNull("b17_feature")) new JSONObject() else
-            JSON.parseObject(record.getString("b17_feature"))
-          val b18: JSONObject = if (record.isNull("b18_feature")) new JSONObject() else
-            JSON.parseObject(record.getString("b18_feature"))
-          val b19: JSONObject = if (record.isNull("b19_feature")) new JSONObject() else
-            JSON.parseObject(record.getString("b19_feature"))
-
-
-          val origin_data = List(
-            (b1, b2, b3, "b123"), (b1, b6, b7, "b167"),
-            (b8, b9, b10, "b8910"), (b11, b12, b13, "b111213"),
-            (b17, b18, b19, "b171819")
-          )
-          for ((b_1, b_2, b_3, prefix1) <- origin_data) {
-            for (prefix2 <- List(
-              "1h", "2h", "3h", "4h", "12h", "1d", "3d", "7d"
-            )) {
-              val exp = if (b_1.isEmpty) 0D else b_1.getIntValue("exp_pv_" + prefix2).toDouble
-              val share = if (b_2.isEmpty) 0D else b_2.getIntValue("share_pv_" + prefix2).toDouble
-              val returns = if (b_3.isEmpty) 0D else b_3.getIntValue("return_uv_" + prefix2).toDouble
-              val f1 = RankExtractorFeature_20240530.calDiv(share, exp)
-              val f2 = RankExtractorFeature_20240530.calLog(share)
-              val f3 = RankExtractorFeature_20240530.calDiv(returns, exp)
-              val f4 = RankExtractorFeature_20240530.calLog(returns)
-              val f5 = f3 * f4
-              val f6 = RankExtractorFeature_20240530.calDiv(returns, share)
-              featureMap.put(prefix1 + "_" + prefix2 + "_" + "STR", f1)
-              featureMap.put(prefix1 + "_" + prefix2 + "_" + "log(share)", f2)
-              featureMap.put(prefix1 + "_" + prefix2 + "_" + "ROV", f3)
-              featureMap.put(prefix1 + "_" + prefix2 + "_" + "log(return)", f4)
-              featureMap.put(prefix1 + "_" + prefix2 + "_" + "ROV*log(return)", f5)
-              featureMap.put(prefix1 + "_" + prefix2 + "_" + "ROS", f6)
-            }
-          }
-
-          val video_info: JSONObject = if (record.isNull("t_v_info_feature")) new JSONObject() else
-            JSON.parseObject(record.getString("t_v_info_feature"))
-          featureMap.put("total_time", if (video_info.containsKey("total_time")) video_info.getIntValue("total_time").toDouble else 0D)
-          featureMap.put("bit_rate", if (video_info.containsKey("bit_rate")) video_info.getIntValue("bit_rate").toDouble else 0D)
-
-          val c1: JSONObject = if (record.isNull("c1_feature")) new JSONObject() else
-            JSON.parseObject(record.getString("c1_feature"))
-          if (c1.nonEmpty) {
-            featureMap.put("playcnt_6h", if (c1.containsKey("playcnt_6h")) c1.getIntValue("playcnt_6h").toDouble else 0D)
-            featureMap.put("playcnt_1d", if (c1.containsKey("playcnt_1d")) c1.getIntValue("playcnt_1d").toDouble else 0D)
-            featureMap.put("playcnt_3d", if (c1.containsKey("playcnt_3d")) c1.getIntValue("playcnt_3d").toDouble else 0D)
-            featureMap.put("playcnt_7d", if (c1.containsKey("playcnt_7d")) c1.getIntValue("playcnt_7d").toDouble else 0D)
-          }
-          val c2: JSONObject = if (record.isNull("c2_feature")) new JSONObject() else
-            JSON.parseObject(record.getString("c2_feature"))
-          if (c2.nonEmpty) {
-            featureMap.put("share_pv_12h", if (c2.containsKey("share_pv_12h")) c2.getIntValue("share_pv_12h").toDouble else 0D)
-            featureMap.put("share_pv_1d", if (c2.containsKey("share_pv_1d")) c2.getIntValue("share_pv_1d").toDouble else 0D)
-            featureMap.put("share_pv_3d", if (c2.containsKey("share_pv_3d")) c2.getIntValue("share_pv_3d").toDouble else 0D)
-            featureMap.put("share_pv_7d", if (c2.containsKey("share_pv_7d")) c2.getIntValue("share_pv_7d").toDouble else 0D)
-            featureMap.put("return_uv_12h", if (c2.containsKey("return_uv_12h")) c2.getIntValue("return_uv_12h").toDouble else 0D)
-            featureMap.put("return_uv_1d", if (c2.containsKey("return_uv_1d")) c2.getIntValue("return_uv_1d").toDouble else 0D)
-            featureMap.put("return_uv_3d", if (c2.containsKey("return_uv_3d")) c2.getIntValue("return_uv_3d").toDouble else 0D)
-            featureMap.put("return_uv_7d", if (c2.containsKey("return_uv_7d")) c2.getIntValue("return_uv_7d").toDouble else 0D)
-          }
-
-          val title = if (video_info.containsKey("title")) video_info.getString("title") else ""
-          if (!title.equals("")) {
-            for (key_feature <- List("c3_feature", "c4_feature", "c5_feature", "c6_feature", "c7_feature")) {
-              val c34567: JSONObject = if (record.isNull(key_feature)) new JSONObject() else
-                JSON.parseObject(record.getString(key_feature))
-              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)
-                  featureMap.put(key_feature + "_" + key_time + "_matchnum", f1)
-                  featureMap.put(key_feature + "_" + key_time + "_maxscore", f3)
-                  featureMap.put(key_feature + "_" + key_time + "_avgscore", f4)
-                }
-              }
-            }
-          }
-
           val vid = if (record.isNull("vid")) "" else record.getString("vid")
-          if (!vid.equals("")) {
-            for (key_feature <- List("c8_feature", "c9_feature")) {
-              val c89: JSONObject = if (record.isNull(key_feature)) new JSONObject() else
-                JSON.parseObject(record.getString(key_feature))
-              for (key_action <- List("share", "return")) {
-                val cfListStr = if (c89.containsKey(key_action)) c89.getString(key_action) else ""
-                if (!cfListStr.equals("")) {
-                  val cfMap = cfListStr.split(",").map(r => {
-                    val rList = r.split(":")
-                    (rList(0), (rList(1), rList(2), rList(3)))
-                  }).toMap
-                  if (cfMap.contains(vid)) {
-                    val (score, num, rank) = cfMap(vid)
-                    featureMap.put(key_feature + "_" + key_action + "_score", score.toDouble)
-                    featureMap.put(key_feature + "_" + key_action + "_num", num.toDouble)
-                    featureMap.put(key_feature + "_" + key_action + "_rank", 1.0 / rank.toDouble)
-                  }
-                }
-              }
-            }
-          }
-
-          val d1: JSONObject = if (record.isNull("d1_feature")) new JSONObject() else
-            JSON.parseObject(record.getString("d1_feature"))
-          if (d1.nonEmpty) {
-            featureMap.put("d1_exp", if (d1.containsKey("exp")) d1.getString("exp").toDouble else 0D)
-            featureMap.put("d1_return_n", if (d1.containsKey("return_n")) d1.getString("return_n").toDouble else 0D)
-            featureMap.put("d1_rovn", if (d1.containsKey("rovn")) d1.getString("rovn").toDouble else 0D)
-          }
-
-
-          /*
-
-
-          视频：
-          曝光使用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 机器型号
-           */
 
+          // a 视频特征
+          val b1: JSONObject = if (record.isNull("b1_feature")) new JSONObject() else JSON.parseObject(record.getString("b1_feature"))
+          val b2: JSONObject = if (record.isNull("b2_feature")) new JSONObject() else JSON.parseObject(record.getString("b2_feature"))
+          val b3: JSONObject = if (record.isNull("b3_feature")) new JSONObject() else JSON.parseObject(record.getString("b3_feature"))
+          val b4: JSONObject = if (record.isNull("b4_feature")) new JSONObject() else JSON.parseObject(record.getString("b3_feature"))
+          val b5: JSONObject = if (record.isNull("b5_feature")) new JSONObject() else JSON.parseObject(record.getString("b3_feature"))
+          val b6: JSONObject = if (record.isNull("b6_feature")) new JSONObject() else JSON.parseObject(record.getString("b6_feature"))
+          val b7: JSONObject = if (record.isNull("b7_feature")) new JSONObject() else JSON.parseObject(record.getString("b7_feature"))
+          val b8: JSONObject = if (record.isNull("b8_feature")) new JSONObject() else JSON.parseObject(record.getString("b8_feature"))
+          val b9: JSONObject = if (record.isNull("b9_feature")) new JSONObject() else JSON.parseObject(record.getString("b9_feature"))
+          val b10: JSONObject = if (record.isNull("b10_feature")) new JSONObject() else JSON.parseObject(record.getString("b10_feature"))
+          val b11: JSONObject = if (record.isNull("b11_feature")) new JSONObject() else JSON.parseObject(record.getString("b11_feature"))
+          val b12: JSONObject = if (record.isNull("b12_feature")) new JSONObject() else JSON.parseObject(record.getString("b12_feature"))
+          val b13: JSONObject = if (record.isNull("b13_feature")) new JSONObject() else JSON.parseObject(record.getString("b13_feature"))
+
+          // 用户特征
+          val c1: JSONObject = if (record.isNull("c1_feature")) new JSONObject() else JSON.parseObject(record.getString("c1_feature"))
+          val c2: JSONObject = if (record.isNull("c2_feature")) new JSONObject() else JSON.parseObject(record.getString("c2_feature"))
+          val c3: JSONObject = if (record.isNull("c3_feature")) new JSONObject() else JSON.parseObject(record.getString("c3_feature"))
+          val c4: JSONObject = if (record.isNull("c4_feature")) new JSONObject() else JSON.parseObject(record.getString("c3_feature"))
+          val c5: JSONObject = if (record.isNull("c5_feature")) new JSONObject() else JSON.parseObject(record.getString("c3_feature"))
+          val c6: JSONObject = if (record.isNull("c6_feature")) new JSONObject() else JSON.parseObject(record.getString("c6_feature"))
+          val c7: JSONObject = if (record.isNull("c7_feature")) new JSONObject() else JSON.parseObject(record.getString("c7_feature"))
+          val c8: JSONObject = if (record.isNull("c8_feature")) new JSONObject() else JSON.parseObject(record.getString("c8_feature"))
+
+          // 视频基础信息 v1-待推荐视频，v2-头部视频
+          val v1: JSONObject = if (record.isNull("v1_feature")) new JSONObject() else JSON.parseObject(record.getString("v1_feature"))
+          val v2: JSONObject = if (record.isNull("v2_feature")) new JSONObject() else JSON.parseObject(record.getString("v2_feature"))
+
+          // CF特征
+          val d1: JSONObject = if (record.isNull("d1_feature")) new JSONObject() else JSON.parseObject(record.getString("d1_feature"))
+          val d2: JSONObject = if (record.isNull("d2_feature")) new JSONObject() else JSON.parseObject(record.getString("d2_feature"))
+          val d3: JSONObject = if (record.isNull("d3_feature")) new JSONObject() else JSON.parseObject(record.getString("d3_feature"))
+
+          val bFeatureMap = new util.HashMap[String, util.Map[String, Object]]();
+          bFeatureMap.put("b1", b1);
+          bFeatureMap.put("b2", b2);
+          bFeatureMap.put("b3", b3);
+          bFeatureMap.put("b4", b4);
+          bFeatureMap.put("b5", b5);
+          bFeatureMap.put("b6", b6);
+          bFeatureMap.put("b7", b7);
+          bFeatureMap.put("b8", b8);
+          bFeatureMap.put("b9", b9);
+          bFeatureMap.put("b10", b10);
+          bFeatureMap.put("b11", b11);
+          bFeatureMap.put("b12", b12);
+          bFeatureMap.put("b13", b13);
+
+          ExtractFeature20250218.handleB1ToB13(bFeatureMap, featureMap);
+          ExtractFeature20250218.handleC1(c1, featureMap)
+          ExtractFeature20250218.handleC2ToC3(c2, c3, featureMap)
+          ExtractFeature20250218.handleC4(c4, featureMap)
+          ExtractFeature20250218.handleC5ToC6(c5, c6, v1, featureMap)
+
+          val c67Map = ExtractFeature20250218.handleC6ToC7(c6, c7)
+          ExtractFeature20250218.useC6ToC7(c67Map, vid, featureMap)
+
+          ExtractFeature20250218.handleD1(d1, featureMap)
+          ExtractFeature20250218.handleD2(d2, featureMap)
+          ExtractFeature20250218.handleD3(d3, featureMap)
+          ExtractFeature20250218.handleVideoBasicFeature(v1, featureMap)
+          ExtractFeature20250218.handleVideoSimilarity(v1, v2, featureMap)
 
           //4 处理label信息。
           val labels = new JSONObject