Merge branch 'feature/jch_makedata' of https://git.yishihui.com/algorithm/recommend-emr-dataprocess into feature/jch_makedata

pom.xml

@@ -181,6 +181,11 @@
             <artifactId>xgboost4j-spark_2.11</artifactId>
             <version>1.1.2</version>
         </dependency>
+        <dependency>
+            <groupId>com.tzld.piaoquan</groupId>
+            <artifactId>recommend-similarity</artifactId>
+            <version>1.0.0</version>
+        </dependency>
     </dependencies>
 
     <build>

src/main/java/examples/utils/PropertiesUtil.java

@@ -0,0 +1,22 @@
+package examples.utils;
+
+import org.springframework.context.EnvironmentAware;
+import org.springframework.core.env.Environment;
+import org.springframework.stereotype.Component;
+
+@Component
+public class PropertiesUtil implements EnvironmentAware {
+
+
+    private static Environment environment;
+
+
+    @Override
+    public void setEnvironment(Environment environment) {
+        this.environment = environment;
+    }
+
+    public static String getString(String name) {
+        return environment.getProperty(name);
+    }
+}

src/main/java/examples/utils/SimilarityUtils.java

@@ -0,0 +1,64 @@
+package examples.utils;
+
+import com.tzld.piaoquan.recommend.similarity.word2vec.Segment;
+import com.tzld.piaoquan.recommend.similarity.word2vec.Word2Vec;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * @author dyp
+ * 从recommend-server复制过来的 @zhangbo 20241211
+ */
+@Slf4j
+public final class SimilarityUtils {
+
+    private static Word2Vec vec = new Word2Vec();
+
+    private static final AtomicBoolean modelLoaded = new AtomicBoolean(false);
+    private static final AtomicBoolean init = new AtomicBoolean(false);
+
+    private static final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
+
+    public static void init() {
+        if (init.compareAndSet(false, true)) {
+            Segment.getWords("1");
+            scheduler.scheduleAtFixedRate(() -> {
+                try {
+                    long start = System.currentTimeMillis();
+                    String endpoint = PropertiesUtil.getString("oss.endpoint");
+                    String bucketName = "art-recommend";
+                    String path = "similarity/word2vec/Google_word2vec_zhwiki210720_300d.bin";
+                    String accessKeyId = "LTAIP6x1l3DXfSxm";
+                    String accessKetSecret = "KbTaM9ars4OX3PMS6Xm7rtxGr1FLon";
+                    Word2Vec temp = new Word2Vec();
+                    temp.loadGoogleModelFromOss(endpoint, bucketName, path, accessKeyId, accessKetSecret);
+                    vec = temp;
+                    long end = System.currentTimeMillis();
+
+                    if (modelLoaded.compareAndSet(false, true)) {
+                        scheduler.shutdown();
+                        log.info("Model loaded successfully cost {}. Scheduled tasks cancelled.", end - start);
+                    }
+
+                } catch (IOException e) {
+                    log.error("loadGoogleModelFromOss error", e);
+                }
+            }, 0, 5, TimeUnit.MINUTES);
+        }
+    }
+
+
+    public static float word2VecSimilarity(String str1, String str2) {
+        List<String> words1 = Segment.getWords(str1);
+        List<String> words2 = Segment.getWords(str2);
+        return vec.sentenceSimilarity(words1, words2);
+    }
+
+
+}

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")), // cate2_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)
@@ -251,31 +252,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 机器型号
            */
 
 
@@ -283,7 +280,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))
@@ -323,8 +320,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 = {
@@ -342,10 +345,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
@@ -357,7 +360,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
     }

src/main/scala/com/aliyun/odps/spark/examples/makedata_recsys_r_rate/makedata_recsys_61_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
 /*
@@ -25,15 +25,16 @@ object makedata_recsys_61_originData_20241209 {
 
     // 1 读取参数
     val param = ParamUtils.parseArgs(args)
+
+    val beginStr = param.getOrElse("beginStr", "2024120912")
+    val endStr = param.getOrElse("endStr", "2024120912")
+    val project = param.getOrElse("project", "loghubods")
+    val table = param.getOrElse("table", "alg_recsys_sample_all_v2")
     val tablePart = param.getOrElse("tablePart", "64").toInt
-    val beginStr = param.getOrElse("beginStr", "2023010100")
-    val endStr = param.getOrElse("endStr", "2023010123")
     val savePath = param.getOrElse("savePath", "/dw/recommend/model/61_origin_data/")
-    val project = param.getOrElse("project", "loghubods")
-    val table = param.getOrElse("table", "XXXX")
     val repartition = param.getOrElse("repartition", "32").toInt
 
-    // 2 读取odps+表信息
+    // 2 odps
     val odpsOps = env.getODPS(sc)
 
     // 3 循环执行数据生产
@@ -49,9 +50,7 @@ object makedata_recsys_61_originData_20241209 {
           transfer = func,
           numPartition = tablePart)
         .map(record => {
-
           val featureMap = new JSONObject()
-
           // a 视频特征
           val b1: JSONObject = if (record.isNull("b1_feature")) new JSONObject() else
             JSON.parseObject(record.getString("b1_feature"))
@@ -145,7 +144,7 @@ object makedata_recsys_61_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)
@@ -187,31 +186,27 @@ object makedata_recsys_61_originData_20241209 {
 
 
           /*
-
-
-          视频：
-          曝光使用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个特征
-
+          视频特征： 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 机器型号
+                    小时 星期 apptype city province pagesource 机器型号
+          总量：     240+2+12+45+12+3 = 314
+          ---------------------------------------------------------------
+          视频特征：
+                  
+
            */
 
 
@@ -258,7 +253,7 @@ object makedata_recsys_61_originData_20241209 {
     record
   }
 
-  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
@@ -270,7 +265,7 @@ object makedata_recsys_61_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
     }