repair redis pipeline 1208

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/Application.java

@@ -29,9 +29,7 @@ import org.springframework.context.annotation.EnableAspectJAutoProxy;
 public class Application {
     public static void main(String[] args) {
         SpringApplication.run(Application.class, args);
-        // UserFeatureToRedisLoader userFeatureToRedisLoader = new UserFeatureToRedisLoader();
-        // userFeatureToRedisLoader.loadFeatureToRedis("user_redis_tmp_smz","20231206");
-        OfflineAdOutSamplesLoader.mutiplyParser("user_video_features_data_final", "20231206");
+        OfflineAdOutSamplesLoader.mutiplyParser("user_video_features_data_final", "20231206", "20231207");
 
     }
 

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/dataloader/FeatureConstructor.java

@@ -54,7 +54,7 @@ public class  FeatureConstructor {
     }
 
 
-    public static TunnelRecordReader loadDataFromOSSSession(String sql, String table, String dt) {
+    public static TunnelRecordReader loadDataFromOSSSession(String sql) {
         Odps odps = new Odps(account);
         odps.setEndpoint(ODPS_CONFIG.get("ENDPOINT"));
         odps.setDefaultProject("loghubods");

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/dataloader/ItemFeatureToRedisLoader.java

@@ -40,7 +40,7 @@ public class ItemFeatureToRedisLoader {
     public void loadFeatureToRedis(String table, String dt) {
         RedisTemplate<String, String> redisTemplate = buildRedisTemplate();
         String sql = String.format("select * from %s where dt ='%s';", table, dt);
-        TunnelRecordReader reader = FeatureConstructor.loadDataFromOSSSession(sql, table, dt);
+        TunnelRecordReader reader = FeatureConstructor.loadDataFromOSSSession(sql);
         Record record;
         Map<String, String> userFeaRedisFormat = new HashMap<String, String>();
         int count = 0;
@@ -54,7 +54,7 @@ public class ItemFeatureToRedisLoader {
                     count++;
                 } else if (count == 200) {
                     redisTemplate.opsForValue().multiSet(userFeaRedisFormat);
-                    System.out.println("------succes add 2000-----");
+                    System.out.println("------succes add 200-----");
                     userFeaRedisFormat = new HashMap<String, String>();
                     count = 0;
                     break;

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/dataloader/OfflineAdOutSamplesLoader.java

@@ -67,9 +67,9 @@ public class OfflineAdOutSamplesLoader {
 
 
     //  主处理逻辑
-    public static void mutiplyParser(String table, String dt) {
-        String sql = String.format("select * from %s where ad_ornot = '0' and apptype != '13' and dt ='%s';", table, dt);
-        TunnelRecordReader reader = FeatureConstructor.loadDataFromOSSSession(sql, table, dt);
+    public static void mutiplyParser(String table, String startDay, String endDay) {
+        String sql = String.format("select * from %s where ad_ornot = '0' and apptype != '13' and dt >='%s' and dt <='%s';", table, startDay, endDay);
+        TunnelRecordReader reader = FeatureConstructor.loadDataFromOSSSession(sql);
         Record record;
         try {
             while ((record = reader.read()) != null) {
@@ -87,7 +87,7 @@ public class OfflineAdOutSamplesLoader {
             System.out.println("--------args 缺失---------");
             return;
         }
-        OfflineAdOutSamplesLoader.mutiplyParser(args[0], args[1]);
+        OfflineAdOutSamplesLoader.mutiplyParser(args[0], args[1], args[1]);
     }
 
 }

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/dataloader/UserFeatureToRedisLoader.java

@@ -42,7 +42,7 @@ public class UserFeatureToRedisLoader {
     public void loadFeatureToRedis(String userTable, String dt) {
         RedisTemplate<String, String> redisTemplate = buildRedisTemplate();
         String sql = String.format("select * from %s where dt ='%s';", userTable, dt);
-        TunnelRecordReader reader = FeatureConstructor.loadDataFromOSSSession(sql, userTable, dt);
+        TunnelRecordReader reader = FeatureConstructor.loadDataFromOSSSession(sql);
         Record record;
         Map<String, String> userFeaRedisFormat = new HashMap<String, String>();
         int count = 0;