feat:添加v2同步逻辑

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

@@ -20,6 +20,7 @@ import org.apache.spark.api.java.JavaSparkContext;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Objects;
 
 /**
  * @author dyp
@@ -58,6 +59,8 @@ public class ODPSToRedis {
             return;
         }
 
+        DTSConfig dtsConfigV2 = dtsConfigService.getV2DTSConfig(argMap);
+
         long count = 0;
         int retry = NumberUtils.toInt(argMap.getOrDefault("retry", "50"), 50);
         while (count <= 0 && retry-- >= 0) {
@@ -127,6 +130,12 @@ public class ODPSToRedis {
             redisService.mSetV2(iterator, config);
         });
 
+        if (Objects.nonNull(dtsConfigV2)){
+            fieldValues.repartition(finalPartationNum).foreachPartition(iterator -> {
+                redisService.mSetV2(iterator, dtsConfigV2);
+            });
+        }
+
         redisService.setMonitor(config, argMap);
 
         jsc.stop();

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

@@ -27,6 +27,7 @@ public class DTSConfig implements Serializable {
         private String prefix;
         private List<String> key;
         private List<String> value;
+        private String featureField;
         private long expire;
     }
 

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

@@ -32,9 +32,16 @@ public class DTSConfigService {
     }
 
     public DTSConfig getDTSConfig(Map<String, String> argMap) {
+        return this.getDTSConfig(argMap, "dts.config");
+    }
+
+    public DTSConfig getV2DTSConfig(Map<String, String> argMap) {
+        return this.getDTSConfig(argMap, "dts.config.v2");
+    }
 
+    private DTSConfig getDTSConfig(Map<String, String> argMap, String apolloConfigKey) {
         List<DTSConfig> dtsConfigs = JSONUtils.fromJson(
-                ConfigService.getAppConfig().getProperty("dts.config", ""),
+                ConfigService.getAppConfig().getProperty(apolloConfigKey, ""),
                 new TypeToken<List<DTSConfig>>() {
                 },
                 Collections.emptyList());

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

@@ -1,5 +1,6 @@
 package com.tzld.piaoquan.recommend.feature.produce.service;
 
+import com.google.common.reflect.TypeToken;
 import com.tzld.piaoquan.recommend.feature.produce.model.DTSConfig;
 import com.tzld.piaoquan.recommend.feature.produce.util.CompressionUtil;
 import com.tzld.piaoquan.recommend.feature.produce.util.JSONUtils;
@@ -11,10 +12,7 @@ import redis.clients.jedis.Pipeline;
 
 import java.io.Serializable;
 import java.nio.charset.StandardCharsets;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -47,7 +45,10 @@ public class RedisService implements Serializable {
             String redisKey = redisKey(record, config);
 
             Map<String, String> valueMap = new HashMap<>();
-            if (CollectionUtils.isNotEmpty(config.getRedis().getValue())) {
+            if (StringUtils.isNotBlank(config.getRedis().getFeatureField())) {
+                record = JSONUtils.fromJson(record.get(config.getRedis().getFeatureField()), new TypeToken<Map<String, String>>() {
+                }, Collections.emptyMap());
+            }else if (CollectionUtils.isNotEmpty(config.getRedis().getValue())) {
                 for (String valueKey : config.getRedis().getValue()) {
                     valueMap.put(valueKey, record.get(valueKey));
                 }
@@ -85,13 +86,6 @@ public class RedisService implements Serializable {
 
         Pipeline pipeline = jedis.pipelined();
         for (Map.Entry<String, String> e : batch.entrySet()) {
-            // pipeline.setex(e.getKey(), expireSeconds, e.getValue());
-            // 压缩后的key，升级后进行替换
-            // try {
-            //     pipeline.setex(e.getKey(), expireSeconds, CompressionUtil.snappyCompress(e.getValue()));
-            // } catch (Exception ex) {
-            // }
-            //
             try {
                 String newKey = String.format("%s:v2", e.getKey());
                 pipeline.setex(newKey.getBytes(StandardCharsets.UTF_8), expireSeconds, CompressionUtil.snappyCompressV2(e.getValue()));