特征生产

recommend-feature-produce/src/main/java/com/tzld/piaoquan/recommend/feature/produce/ODPSToRedis.java

@@ -34,7 +34,7 @@ public class ODPSToRedis {
         }
 
         SparkConf sparkConf = new SparkConf()
-                .setMaster("local")
+                // .setMaster("local")
                 .setAppName("odps sync to redis");
         for (Map.Entry<String, String> e : argMap.entrySet()) {
             sparkConf.set(e.getKey(), e.getValue());
@@ -70,6 +70,7 @@ public class ODPSToRedis {
         }
 
         log.info("odps count {}", count);
+        argMap.put("partitionNum", String.valueOf(count / 100000 + 1));
 
         JavaRDD<Map<String, String>> fieldValues = odpsService.read(jsc, config, argMap);
         if (fieldValues == null) {

recommend-feature-produce/src/main/java/com/tzld/piaoquan/recommend/feature/produce/Test.java

@@ -10,15 +10,13 @@ public class Test {
         args[0] = "-project";
         args[1] = "loghubods";
         args[2] = "-table";
-        args[3] = "alg_vid_feature_all_exp";
+        args[3] = "alg_mid_feature_play";
         args[4] = "-dt";
         args[5] = "20240612";
         args[6] = "-hh";
         args[7] = "15";
         args[8] = "-env";
         args[9] = "test";
-        args[10] = "-partitionNum";
-        args[11] = "50";
         ODPSToRedis.main(args);
     }
 

recommend-feature-produce/src/main/java/com/tzld/piaoquan/recommend/feature/produce/service/RedisService.java

@@ -61,7 +61,7 @@ public class RedisService implements Serializable {
             String redisKey = redisKey(record, config);
             String value = JSONUtils.toJson(record);
             batch.put(redisKey, value);
-            if (batch.size() > 1000) {
+            if (batch.size() > 2000) {
                 mSet(batch, config.getRedis().getExpire(), TimeUnit.SECONDS);
                 batch.clear();
             }