Merge branch 'feature_20260617_redis_store_optimize_v3' of algorithm/recommend-feature into master

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) {
@@ -123,9 +126,12 @@ public class ODPSToRedis {
         int finalPartationNum = Math.min(maxPartitionNum, calcPartationNum);
 
         log.info("argMap.maxPartitionNum: {} calcPartationNum: {}, finalPartationNum: {}", maxPartitionNum, calcPartationNum, finalPartationNum);
-        fieldValues.repartition(finalPartationNum).foreachPartition(iterator -> {
-            redisService.mSetV2(iterator, config);
-        });
+        log.info("config: {}, dtsConfigV2: {}", config, dtsConfigV2);
+        JavaRDD<Map<String, String>> repartitionedFieldValues = fieldValues.repartition(finalPartationNum);
+        repartitionedFieldValues.foreachPartition(iterator -> redisService.mSetV2(iterator, config, false));
+        if (Objects.nonNull(dtsConfigV2) && dtsConfigV2.selfCheck()) {
+            repartitionedFieldValues.foreachPartition(iterator -> redisService.mSetV2(iterator, dtsConfigV2, true));
+        }
 
         redisService.setMonitor(config, argMap);
 

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,14 +32,21 @@ 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());
         if (CollectionUtils.isEmpty(dtsConfigs)) {
-            log.error("DTSConfig not config");
+            log.error("DTSConfig not config, apolloConfigKey: {}", apolloConfigKey);
             return null;
         }
         Optional<DTSConfig> optional = dtsConfigs.stream()

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

@@ -1,8 +1,10 @@
 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;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang3.StringUtils;
@@ -11,10 +13,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;
 
 /**
@@ -22,6 +21,7 @@ import java.util.concurrent.TimeUnit;
  *
  * @author dyp
  */
+@Slf4j
 public class RedisService implements Serializable {
     private int port = 6379;
     private String password = "";
@@ -37,7 +37,8 @@ public class RedisService implements Serializable {
         }
     }
 
-    public void mSetV2(Iterator<Map<String, String>> dataIte, DTSConfig config) {
+    public void mSetV2(Iterator<Map<String, String>> dataIte, DTSConfig config, boolean isV3Optimize) {
+        log.info("mSetV2 param: {}, isV3Optimize: {}", config, isV3Optimize);
         Map<String, String> batch = new HashMap<>();
         Jedis jedis = new Jedis(hostName, port);
         jedis.auth(password);
@@ -47,7 +48,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));
                 }
@@ -57,7 +61,7 @@ public class RedisService implements Serializable {
             String value = JSONUtils.toJson(record);
             batch.put(redisKey, value);
             if (batch.size() >= 500) {
-                mSet(jedis, batch, expire, TimeUnit.SECONDS);
+                mSet(jedis, batch, expire, TimeUnit.SECONDS, isV3Optimize);
                 batch.clear();
                 try {
                     TimeUnit.MILLISECONDS.sleep(10);
@@ -67,7 +71,7 @@ public class RedisService implements Serializable {
             }
         }
         if (MapUtils.isNotEmpty(batch)) {
-            mSet(jedis, batch, expire, TimeUnit.SECONDS);
+            mSet(jedis, batch, expire, TimeUnit.SECONDS, isV3Optimize);
         }
         jedis.close();
     }
@@ -80,21 +84,17 @@ public class RedisService implements Serializable {
         jedis.setex(redisKey, expire, "");
     }
 
-    private void mSet(Jedis jedis, Map<String, String> batch, long expire, TimeUnit timeUnit) {
+    private void mSet(Jedis jedis, Map<String, String> batch, long expire, TimeUnit timeUnit, boolean isV3Optimize) {
         long expireSeconds = timeUnit.toSeconds(expire);
 
         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()));
+                String key = e.getKey();
+                if (!isV3Optimize) {
+                    key = String.format("%s:v2", e.getKey());
+                }
+                pipeline.setex(key.getBytes(StandardCharsets.UTF_8), expireSeconds, CompressionUtil.snappyCompressV2(e.getValue()));
             } catch (Exception ignore) {
             }
         }

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

@@ -43,94 +43,36 @@ public class FeatureV2Service {
     @ApolloJsonValue("${dts.config:}")
     private List<DTSConfig> newDtsConfigs;
 
+    @ApolloJsonValue("${dts.config.v2:}")
+    private List<DTSConfig> dtsConfigsV2;
+
     @Value("${feature.mget.batch.size:300}")
     private Integer batchSize;
 
-    @Value("${remove.base64.switch:false}")
-    private Boolean removeBase64Switch;
+    @Value("${feature.service.optimize.switch:false}")
+    private Boolean optimizeV3Switch;
 
     public MultiGetFeatureResponse multiGetFeature(MultiGetFeatureRequest request) {
         int keyCount = request.getFeatureKeyCount();
-
         if (keyCount == 0) {
             return MultiGetFeatureResponse.newBuilder()
                     .setResult(Result.newBuilder().setCode(1))
                     .build();
         }
 
-
-        // 移除Base64编解码的key
-        if (Boolean.TRUE.equals(removeBase64Switch)) {
-            return this.multiGetFeatureV2(request);
-        }
-
         long startTime = System.currentTimeMillis();
 
-        // 1. 生成Redis Key列表
-        List<String> redisKeys = CommonCollectionUtils.toList(request.getFeatureKeyList(), this::redisKey);
-
-        int safeBatchSize = (batchSize == null || batchSize <= 0) ? 300 : batchSize;
-
-        // 2. 分批并行查询（必须保证返回值与 key 顺序一致）
-        List<BatchFuture<String>> futures = new ArrayList<>();
-        for (int start = 0; start < redisKeys.size(); start += safeBatchSize) {
-            int end = Math.min(start + safeBatchSize, redisKeys.size());
-            List<String> batchKeys = redisKeys.subList(start, end);
-            Future<List<String>> future = ThreadPoolFactory.multiGetFeaturePool()
-                    .submit(() -> redisTemplate.opsForValue().multiGet(batchKeys));
-            futures.add(new BatchFuture<>(start, end - start, future));
-        }
-
-        // 3. 收集结果（整体最多等待约 2s；超时/异常的批次用 null 填充，避免错位）
-        List<String> allValues = new ArrayList<>(Collections.nCopies(keyCount, null));
-        long deadlineNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(2000);
-        for (BatchFuture<String> bf : futures) {
-            long remainingNanos = deadlineNanos - System.nanoTime();
-            if (remainingNanos <= 0) {
-                bf.future.cancel(true);
-                continue;
-            }
-            try {
-                List<String> batchValues = bf.future.get(remainingNanos, TimeUnit.NANOSECONDS);
-                if (batchValues == null) {
-                    continue;
-                }
-                int copyLen = Math.min(bf.size, batchValues.size());
-                for (int i = 0; i < copyLen; i++) {
-                    allValues.set(bf.startIndex + i, batchValues.get(i));
-                }
-            } catch (TimeoutException e) {
-                bf.future.cancel(true);
-                log.warn("Batch mGet timeout, startIndex={}, size={}", bf.startIndex, bf.size);
-            } catch (Exception e) {
-                log.error("Batch mGet failed, startIndex={}, size={}", bf.startIndex, bf.size, e);
-            }
-        }
-
-        // 4. 解压缩
-        List<String> values = CommonCollectionUtils.toList(allValues, CompressionUtil::snappyDecompress);
-
-        // 5. 构建响应
-        MultiGetFeatureResponse.Builder builder = MultiGetFeatureResponse.newBuilder();
-        builder.setResult(Result.newBuilder().setCode(1));
-        for (int i = 0; i < keyCount; i++) {
-            builder.putFeature(request.getFeatureKeyList().get(i).getUniqueKey(),
-                    Strings.nullToEmpty(values.get(i)));
+        List<String> redisKeys;
+        if (Boolean.TRUE.equals(optimizeV3Switch)) {
+            redisKeys = request.getFeatureKeyList().stream()
+                    .map(this::buildRedisKeyV2)
+                    .collect(Collectors.toList());
+        } else {
+            redisKeys = request.getFeatureKeyList().stream()
+                    .map(this::redisKey)
+                    .map(s -> String.format("%s:v2", s))
+                    .collect(Collectors.toList());
         }
-        MultiGetFeatureResponse build = builder.build();
-        log.info("multiGetFeature, cost={}ms, keyCount={}", System.currentTimeMillis() - startTime, keyCount);
-        return build;
-    }
-
-    private MultiGetFeatureResponse multiGetFeatureV2(MultiGetFeatureRequest request) {
-        long startTime = System.currentTimeMillis();
-        int keyCount = request.getFeatureKeyCount();
-
-        // 1. 生成Redis Key列表, v2去除了base64编码 key格式也变了
-        List<String> redisKeys = request.getFeatureKeyList().stream()
-                .map(this::redisKey)
-                .map(s -> String.format("%s:v2", s))
-                .collect(Collectors.toList());
 
         int safeBatchSize = (batchSize == null || batchSize <= 0) ? 300 : batchSize;
 
@@ -181,7 +123,7 @@ public class FeatureV2Service {
                     Strings.nullToEmpty(values.get(i)));
         }
         MultiGetFeatureResponse build = builder.build();
-        log.info("multiGetFeatureV2, cost={}ms, keyCount={}", System.currentTimeMillis() - startTime, keyCount);
+        log.info("multiGetFeature, cost={}ms, keyCount={}", System.currentTimeMillis() - startTime, keyCount);
         return build;
     }
 
@@ -197,10 +139,17 @@ public class FeatureV2Service {
         }
     }
 
-    // Note：写入和读取的key生成规则应保持一致
     private String redisKey(FeatureKeyProto fk) {
+        return this.buildRedisKey(fk, newDtsConfigs);
+    }
 
-        Optional<DTSConfig> optional = newDtsConfigs.stream()
+    private String buildRedisKeyV2(FeatureKeyProto fk) {
+        return this.buildRedisKey(fk, dtsConfigsV2);
+    }
+
+    // Note：写入和读取的key生成规则应保持一致
+    private String buildRedisKey(FeatureKeyProto fk, List<DTSConfig> dtsConfigs){
+        Optional<DTSConfig> optional = dtsConfigs.stream()
                 .filter(c -> c.getOdps() != null && StringUtils.equals(c.getOdps().getTable(), fk.getTableName()))
                 .findFirst();
         if (!optional.isPresent()) {
@@ -224,5 +173,4 @@ public class FeatureV2Service {
         }
         return sb.toString();
     }
-
 }

recommend-feature-service/src/main/resources/application.yml

@@ -18,5 +18,5 @@ app:
 apollo:
   bootstrap:
     enabled: true
-    namespaces: application
+    namespaces: application,RD.Experiment
   cacheDir: /datalog/apollo-cache-dir