feat:同步任务抽取抽象类

recommend-feature-common/pom.xml

@@ -12,6 +12,14 @@
     <artifactId>recommend-feature-common</artifactId>
 
     <dependencies>
+
+        <!-- 阿里云服务SDK -->
+        <dependency>
+            <groupId>com.aliyun.openservices</groupId>
+            <artifactId>tablestore</artifactId>
+            <version>5.17.7</version>
+        </dependency>
+
         <!-- 工具类依赖 -->
         <dependency>
             <groupId>org.projectlombok</groupId>

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

@@ -1,6 +1,7 @@
 package com.tzld.piaoquan.recommend.feature.common.model;
 
 import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
 
 import java.io.Serializable;
 import java.util.List;
@@ -8,10 +9,12 @@ import java.util.List;
 /**
  * @author dyp
  */
+@Slf4j
 @Data
 public class DTSConfig implements Serializable {
     private ODPS odps;
     private Redis redis;
+    private TableStore tableStore;
 
     @Data
     public static class ODPS implements Serializable {
@@ -29,13 +32,38 @@ public class DTSConfig implements Serializable {
         private long expire;
     }
 
+    @Data
+    public static class TableStore implements Serializable {
+        private String compressMode;
+        private String keyVersion;
+        private String prefix;
+        private List<String> key;
+        private List<String> value;
+        private String featureField;
+    }
+
     public boolean selfCheck() {
         if (odps == null) {
+            log.error("odps is empty");
             return false;
         }
         if (redis == null) {
+            log.error("redis is empty");
+            return false;
+        }
+        return true;
+    }
+
+    public boolean tableStoreSelfCheck(){
+        if (odps == null) {
+            log.error("odps is empty");
+            return false;
+        }
+        if (tableStore == null) {
+            log.error("tableStore is empty");
             return false;
         }
         return true;
     }
+
 }

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

@@ -0,0 +1,86 @@
+package com.tzld.piaoquan.recommend.feature.common.service;
+
+import com.alicloud.openservices.tablestore.SyncClient;
+import com.alicloud.openservices.tablestore.core.ResourceManager;
+import com.alicloud.openservices.tablestore.core.auth.CredentialsProvider;
+import com.alicloud.openservices.tablestore.core.auth.DefaultCredentialProvider;
+import com.alicloud.openservices.tablestore.core.auth.DefaultCredentials;
+import com.alicloud.openservices.tablestore.core.auth.V4Credentials;
+import com.alicloud.openservices.tablestore.model.*;
+import com.tzld.piaoquan.recommend.feature.common.model.DTSConfig;
+import com.tzld.piaoquan.recommend.feature.common.util.CompressionUtil;
+import com.tzld.piaoquan.recommend.feature.common.util.TableStoreUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.MapUtils;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+
+@Slf4j
+public class TableStoreService {
+    private static final String accessId = "LTAI5t5jJUZPSHCsKMHrnFFa";
+    private static final String accessKey = "2DdLqVtVBupG1JnZ9rdDJvC6Jf2NzQ";
+
+    private static final String region = "cn-hangzhou";
+    private static final String instanceName = "algo-feature";
+    private static final String endpoint = "https://algo-feature.cn-hangzhou.vpc.tablestore.aliyuncs.com";
+
+    private final SyncClient client;
+
+    public TableStoreService() {
+        // 构造凭证
+        DefaultCredentials credentials = new DefaultCredentials(accessId, accessKey);
+        V4Credentials credentialsV4 = V4Credentials.createByServiceCredentials(credentials, region);
+        CredentialsProvider provider = new DefaultCredentialProvider(credentialsV4);
+
+        // 创建客户端实例
+        this.client = new SyncClient(endpoint, provider, instanceName, null, new ResourceManager(null, null));
+
+    }
+
+    public void batchWrite(Iterator<Map<String, String>> record, DTSConfig dtsConfig) {
+        Map<String, String> batch = new HashMap<>();
+        while (record.hasNext()) {
+            Map<String, String> recordMap = record.next();
+
+            String key = TableStoreUtil.buildKey(recordMap, dtsConfig);
+            String valueMap = TableStoreUtil.buildValue(recordMap, dtsConfig);
+
+            batch.put(key, valueMap);
+
+            if (batch.size() >= 200) {
+                this.batchWrite(batch);
+                batch.clear();
+            }
+        }
+
+        if (MapUtils.isNotEmpty(batch)) {
+            this.batchWrite(batch);
+        }
+
+    }
+
+    private void batchWrite(Map<String, String> data) {
+        log.info("batchWrite data.size: {}", data.size());
+        try {
+
+            BatchWriteRowRequest request = new BatchWriteRowRequest();
+
+            for (Map.Entry<String, String> entry : data.entrySet()) {
+                PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
+                primaryKeyBuilder.addPrimaryKeyColumn("key", PrimaryKeyValue.fromString(entry.getKey()));
+
+                RowPutChange rowPutChange = new RowPutChange("feature_prod", primaryKeyBuilder.build());
+                rowPutChange.addColumn("feature", ColumnValue.fromBinary(CompressionUtil.snappyCompressV2(entry.getValue())));
+
+                request.addRowChange(rowPutChange);
+            }
+
+            client.batchWriteRow(request);
+
+        } catch (Exception e) {
+            log.error("", e);
+        }
+    }
+}

recommend-feature-common/src/main/java/com/tzld/piaoquan/recommend/feature/common/util/TableStoreUtil.java

@@ -0,0 +1,49 @@
+package com.tzld.piaoquan.recommend.feature.common.util;
+
+import com.google.common.reflect.TypeToken;
+import com.tzld.piaoquan.recommend.feature.common.model.DTSConfig;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class TableStoreUtil {
+
+    public static String buildKey(Map<String, String> recordMap, DTSConfig dtsConfig) {
+
+        DTSConfig.TableStore tableStore = dtsConfig.getTableStore();
+
+        StringBuilder builder = new StringBuilder();
+        builder.append(tableStore.getCompressMode());
+        builder.append(":");
+        builder.append(tableStore.getKeyVersion());
+        builder.append(":");
+        builder.append(tableStore.getPrefix());
+        if (CollectionUtils.isNotEmpty(tableStore.getKey())) {
+            for (String item : tableStore.getKey()) {
+                builder.append(":");
+                builder.append(recordMap.get(item));
+            }
+        }
+
+        return builder.toString();
+    }
+
+    public static String buildValue(Map<String, String> recordMap, DTSConfig config) {
+        Map<String, String> valueMap = new HashMap<>();
+        DTSConfig.TableStore tableStore = config.getTableStore();
+        if (StringUtils.isNotBlank(tableStore.getFeatureField())) {
+            valueMap = JSONUtils.fromJson(recordMap.get(tableStore.getFeatureField()), new TypeToken<Map<String, String>>() {
+            }, Collections.emptyMap());
+        } else if (CollectionUtils.isNotEmpty(tableStore.getValue())) {
+            for (String valueKey : tableStore.getValue()) {
+                valueMap.put(valueKey, recordMap.get(valueKey));
+            }
+        } else {
+            valueMap = recordMap;
+        }
+        return JSONUtils.toJson(valueMap);
+    }
+}

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

@@ -0,0 +1,157 @@
+package com.tzld.piaoquan.recommend.feature.produce;
+
+import com.alibaba.fastjson.JSONObject;
+import com.squareup.okhttp.MediaType;
+import com.squareup.okhttp.OkHttpClient;
+import com.squareup.okhttp.Request;
+import com.squareup.okhttp.RequestBody;
+import com.tzld.piaoquan.recommend.feature.common.model.DTSConfig;
+import com.tzld.piaoquan.recommend.feature.common.util.JSONUtils;
+import com.tzld.piaoquan.recommend.feature.produce.service.CMDService;
+import com.tzld.piaoquan.recommend.feature.produce.service.DTSConfigService;
+import com.tzld.piaoquan.recommend.feature.produce.service.ODPSService;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang3.math.NumberUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Slf4j
+public abstract class AbstractODPSSyncJob {
+
+    protected final static CMDService cmdService = new CMDService();
+    protected final static ODPSService odpsService = new ODPSService();
+
+    /**
+     * 模板方法：固化 ODPS 数据同步到下游存储的公共编排流程。
+     * 子类只需实现 appName / loadConfig / validate / sink，并按需重写 afterSink。
+     */
+    protected void run(String[] args) {
+        log.info("args {}", JSONUtils.toJson(args));
+
+        Map<String, String> argMap = cmdService.parse(args);
+        if (MapUtils.isEmpty(argMap)) {
+            log.error("args is empty");
+            return;
+        }
+
+        SparkConf sparkConf = new SparkConf()
+                //.setMaster("local")
+                .setAppName(appName(argMap));
+        for (Map.Entry<String, String> e : argMap.entrySet()) {
+            sparkConf.set(e.getKey(), e.getValue());
+        }
+        JavaSparkContext jsc = new JavaSparkContext(sparkConf);
+
+        // ODPS
+        log.info("read odps");
+        String env = argMap.get("env");
+        DTSConfigService dtsConfigService = new DTSConfigService(env);
+        DTSConfig config = loadConfig(dtsConfigService, argMap);
+        if (config == null || !validate(config)) {
+            log.error("dts config error");
+            return;
+        }
+
+        Pair<Long, JavaRDD<Map<String, String>>> pair = readODPS(jsc, config, argMap);
+        Long count = pair.getKey();
+        JavaRDD<Map<String, String>> records = pair.getValue();
+
+        if (records == null) {
+            log.info("odps empty");
+            return;
+        }
+
+        log.info("odps count {}", count);
+
+        long odpsBatchSize = NumberUtils.toLong(argMap.getOrDefault("odpsBatchSize", "1000000"));
+        int calcPartitionNum = (int) (count / odpsBatchSize + 1);
+
+        int maxPartitionNum = NumberUtils.toInt(argMap.get("maxPartitionNum"), 30);
+        int finalPartitionNum = Math.min(maxPartitionNum, calcPartitionNum);
+
+        log.info("argMap.maxPartitionNum: {} calcPartitionNum: {}, finalPartitionNum: {}", maxPartitionNum, calcPartitionNum, finalPartitionNum);
+        log.info("config: {}", config);
+
+        // RDD
+        JavaRDD<Map<String, String>> repartitionedRdd = records.repartition(finalPartitionNum);
+        sink(repartitionedRdd, config, argMap);
+
+        afterSink(config, argMap);
+
+        jsc.stop();
+    }
+
+    protected abstract String appName(Map<String, String> argMap);
+
+    protected abstract DTSConfig loadConfig(DTSConfigService dtsConfigService, Map<String, String> argMap);
+
+    protected abstract boolean validate(DTSConfig config);
+
+    protected abstract void sink(JavaRDD<Map<String, String>> repartitionedRdd, DTSConfig config, Map<String, String> argMap);
+
+    /**
+     * 同步完成后的收尾钩子（如设置监控标记），默认空实现，子类按需重写。
+     */
+    protected void afterSink(DTSConfig config, Map<String, String> argMap) {
+    }
+
+    protected static Pair<Long, JavaRDD<Map<String, String>>> readODPS(JavaSparkContext jsc, DTSConfig config, Map<String, String> argMap) {
+
+        long count = 0;
+        int retry = NumberUtils.toInt(argMap.getOrDefault("retry", "50"), 50);
+        while (count <= 0 && retry-- >= 0) {
+            count = odpsService.count(config, argMap);
+            if (count <= 0) {
+                if (retry <= 0) {
+                    sendFeishuAlert(argMap);
+                    Map<String, String> partitionMap = odpsService.getLastestPartition(config, argMap);
+                    argMap.putAll(partitionMap);
+                    log.info("partitionMap {} argMap {}", JSONUtils.toJson(partitionMap), JSONUtils.toJson(argMap));
+                    count = odpsService.count(config, argMap);
+                    if (count <= 0) {
+                        return Pair.of(count, null);
+                    }
+                } else {
+                    try {
+                        log.info("sleep {}", retry);
+                        Thread.sleep(60000);
+                    } catch (InterruptedException e) {
+                        log.error("sleep error ", e);
+                    }
+                }
+            }
+        }
+
+        JavaRDD<Map<String, String>> records = odpsService.read(jsc, config, argMap);
+        return Pair.of(count, records);
+    }
+
+    private static void sendFeishuAlert(Map<String, String> argMap) {
+        try {
+            // TODO 报警
+            String feishuWebhook = "https://open.feishu.cn/open-apis/bot/v2/hook/540d4098-367a-4068-9a44-b8109652f07c";
+            MediaType mediaType = MediaType.parse("application/json");
+            JSONObject param = new JSONObject();
+            param.put("msg_type", "text");
+            Map<String, String> map = new HashMap<>();
+            map.put("text", argMap.get("table") + " 使用兜底数据");
+            param.put("content", map);
+            RequestBody body = RequestBody.create(mediaType, param.toJSONString());
+            Request request = new Request.Builder()
+                    .url(feishuWebhook)
+                    .method("POST", body)
+                    .addHeader("Content-Type", "application/json")
+                    .build();
+            OkHttpClient client = new OkHttpClient();
+            client.newCall(request).execute();
+        } catch (Throwable e) {
+            log.error("send feishu alert error ", e);
+        }
+    }
+}

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

@@ -1,83 +0,0 @@
-package com.tzld.piaoquan.recommend.feature.produce;
-
-import com.tzld.piaoquan.recommend.feature.common.model.DTSConfig;
-import com.tzld.piaoquan.recommend.feature.produce.service.CMDService;
-import com.tzld.piaoquan.recommend.feature.produce.service.DTSConfigService;
-import com.tzld.piaoquan.recommend.feature.produce.service.ODPSService;
-import com.tzld.piaoquan.recommend.feature.produce.service.RedisService;
-import com.tzld.piaoquan.recommend.feature.common.util.JSONUtils;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.collections.MapUtils;
-import org.apache.commons.lang3.math.NumberUtils;
-import org.apache.spark.SparkConf;
-import org.apache.spark.api.java.JavaRDD;
-import org.apache.spark.api.java.JavaSparkContext;
-
-import java.util.Map;
-
-@Slf4j
-public class FeatureRedisClean {
-
-    private static CMDService cmdService = new CMDService();
-    private static ODPSService odpsService = new ODPSService();
-
-    public static void main(String[] args) {
-        log.info("args {}", JSONUtils.toJson(args));
-
-        Map<String, String> argMap = cmdService.parse(args);
-
-        if (MapUtils.isEmpty(argMap)) {
-            log.error("args is empty");
-            return;
-        }
-
-        SparkConf sparkConf = new SparkConf()
-                .setAppName("odps sync to redis : " + argMap.get("table"));
-        for (Map.Entry<String, String> e : argMap.entrySet()) {
-            sparkConf.set(e.getKey(), e.getValue());
-        }
-        JavaSparkContext jsc = new JavaSparkContext(sparkConf);
-
-
-        // ODPS
-        log.info("read odps");
-        String env = argMap.get("env");
-        DTSConfigService dtsConfigService = new DTSConfigService(env);
-        DTSConfig config = dtsConfigService.getDTSConfig(argMap);
-        if (config == null || !config.selfCheck()) {
-            log.error("dts config error");
-            return;
-        }
-
-        long count = odpsService.count(config, argMap);
-        if (count == 0) {
-            log.error("table {} not data", argMap.get("table"));
-            return;
-        }
-
-        JavaRDD<Map<String, String>> fieldValues = odpsService.read(jsc, config, argMap);
-        if (fieldValues == null) {
-            log.info("odps empty");
-            return;
-        }
-
-        // RDD
-        log.info("sync redis");
-        RedisService redisService = new RedisService(env);
-
-
-        log.info("odps count {}", count);
-        long odpsBatchSize = NumberUtils.toLong(argMap.getOrDefault("odpsBatchSize", "200000"));
-        int calcPartationNum = (int) (count / odpsBatchSize + 1);
-
-        int maxPartitionNum = NumberUtils.toInt(argMap.get("maxPartitionNum"), 30);
-        int finalPartationNum = Math.min(maxPartitionNum, calcPartationNum);
-
-        log.info("argMap.maxPartitionNum: {} calcPartationNum: {}, finalPartationNum: {}", maxPartitionNum, calcPartationNum, finalPartationNum);
-        log.info("config: {}", config);
-        JavaRDD<Map<String, String>> repartitionedFieldValues = fieldValues.repartition(finalPartationNum);
-        repartitionedFieldValues.foreachPartition(iterator -> redisService.delete(iterator, config));
-
-        jsc.stop();
-    }
-}

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

@@ -1,135 +1,48 @@
 package com.tzld.piaoquan.recommend.feature.produce;
 
-import com.alibaba.fastjson.JSONObject;
-import com.squareup.okhttp.MediaType;
-import com.squareup.okhttp.OkHttpClient;
-import com.squareup.okhttp.Request;
-import com.squareup.okhttp.RequestBody;
 import com.tzld.piaoquan.recommend.feature.common.model.DTSConfig;
-import com.tzld.piaoquan.recommend.feature.produce.service.CMDService;
 import com.tzld.piaoquan.recommend.feature.produce.service.DTSConfigService;
-import com.tzld.piaoquan.recommend.feature.produce.service.ODPSService;
 import com.tzld.piaoquan.recommend.feature.produce.service.RedisService;
-import com.tzld.piaoquan.recommend.feature.common.util.JSONUtils;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.collections.MapUtils;
-import org.apache.commons.lang3.math.NumberUtils;
-import org.apache.spark.SparkConf;
 import org.apache.spark.api.java.JavaRDD;
-import org.apache.spark.api.java.JavaSparkContext;
 
-import java.util.HashMap;
 import java.util.Map;
 
 /**
  * @author dyp
  */
 @Slf4j
-public class ODPSToRedis {
-    private static CMDService cmdService = new CMDService();
-    private static ODPSService odpsService = new ODPSService();
+public class ODPSToRedis extends AbstractODPSSyncJob {
 
     public static void main(String[] args) {
+        new ODPSToRedis().run(args);
+    }
 
-        log.info("args {}", JSONUtils.toJson(args));
-
-        Map<String, String> argMap = cmdService.parse(args);
-
-        if (MapUtils.isEmpty(argMap)) {
-            log.error("args is empty");
-            return;
-        }
-
-        SparkConf sparkConf = new SparkConf()
-                //.setMaster("local")
-                .setAppName("odps sync to redis : " + argMap.get("table"));
-        for (Map.Entry<String, String> e : argMap.entrySet()) {
-            sparkConf.set(e.getKey(), e.getValue());
-        }
-        JavaSparkContext jsc = new JavaSparkContext(sparkConf);
-
-        // ODPS
-        log.info("read odps");
-        String env = argMap.get("env");
-        DTSConfigService dtsConfigService = new DTSConfigService(env);
-        DTSConfig config = dtsConfigService.getDTSConfig(argMap);
-        if (config == null || !config.selfCheck()) {
-            log.error("dts config error");
-            return;
-        }
+    @Override
+    protected String appName(Map<String, String> argMap) {
+        return "odps sync to redis : " + argMap.get("table");
+    }
 
-        long count = 0;
-        int retry = NumberUtils.toInt(argMap.getOrDefault("retry", "50"), 50);
-        while (count <= 0 && retry-- >= 0) {
-            count = odpsService.count(config, argMap);
-            if (count <= 0) {
-                if (retry <= 0) {
-                    try {
-                        // TODO 报警
-                        String feishuWebhook = "https://open.feishu.cn/open-apis/bot/v2/hook/540d4098-367a-4068-9a44" +
-                                "-b8109652f07c";
-                        MediaType mediaType = MediaType.parse("application/json");
-                        JSONObject param = new JSONObject();
-                        param.put("msg_type", "text");
-                        Map<String, String> map = new HashMap<>();
-                        map.put("text", argMap.get("table") + " 使用兜底数据");
-                        param.put("content", map);
-                        RequestBody body = RequestBody.create(mediaType, param.toJSONString());
-                        Request request = new Request.Builder()
-                                .url(feishuWebhook)
-                                .method("POST", body)
-                                .addHeader("Content-Type", "application/json")
-                                .build();
-                        OkHttpClient client = new OkHttpClient();
-                        client.newCall(request).execute();
-                    } catch (Throwable e) {
-                        log.error("send feishu alert error ", e);
-                    }
-                    Map<String, String> partitionMap = odpsService.getLastestPartition(config, argMap);
-                    argMap.putAll(partitionMap);
-                    log.info("partitionMap {} argMap {}", JSONUtils.toJson(partitionMap), JSONUtils.toJson(argMap));
-                    count = odpsService.count(config, argMap);
-                    if (count <= 0) {
-                        return;
-                    }
-                } else {
-                    try {
-                        log.info("sleep {}", retry);
-                        Thread.sleep(60000);
-                    } catch (InterruptedException e) {
-                        e.printStackTrace();
-                    }
-                }
-            }
-        }
+    @Override
+    protected DTSConfig loadConfig(DTSConfigService dtsConfigService, Map<String, String> argMap) {
+        return dtsConfigService.getDTSConfig(argMap);
+    }
 
-        JavaRDD<Map<String, String>> fieldValues = odpsService.read(jsc, config, argMap);
-        if (fieldValues == null) {
-            log.info("odps empty");
-            return;
-        }
+    @Override
+    protected boolean validate(DTSConfig config) {
+        return config.selfCheck();
+    }
 
-        // RDD
+    @Override
+    protected void sink(JavaRDD<Map<String, String>> repartitionedRdd, DTSConfig config, Map<String, String> argMap) {
         log.info("sync redis");
-        RedisService redisService = new RedisService(env);
-
-
-        log.info("odps count {}", count);
-
-        long odpsBatchSize = NumberUtils.toLong(argMap.getOrDefault("odpsBatchSize", "200000"));
-        int calcPartationNum = (int) (count / odpsBatchSize + 1);
-
-        int maxPartitionNum = NumberUtils.toInt(argMap.get("maxPartitionNum"), 30);
-        int finalPartationNum = Math.min(maxPartitionNum, calcPartationNum);
-
-        log.info("argMap.maxPartitionNum: {} calcPartationNum: {}, finalPartationNum: {}", maxPartitionNum, calcPartationNum, finalPartationNum);
-        log.info("config: {}", config);
-        JavaRDD<Map<String, String>> repartitionedFieldValues = fieldValues.repartition(finalPartationNum);
-        repartitionedFieldValues.foreachPartition(iterator -> redisService.mSetV2(iterator, config));
-
-        redisService.setMonitor(config, argMap);
+        RedisService redisService = new RedisService(argMap.get("env"));
+        repartitionedRdd.foreachPartition(iterator -> redisService.mSetV2(iterator, config));
+    }
 
-        jsc.stop();
+    @Override
+    protected void afterSink(DTSConfig config, Map<String, String> argMap) {
+        new RedisService(argMap.get("env")).setMonitor(config, argMap);
     }
 
 }

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

@@ -0,0 +1,41 @@
+package com.tzld.piaoquan.recommend.feature.produce;
+
+import com.tzld.piaoquan.recommend.feature.common.model.DTSConfig;
+import com.tzld.piaoquan.recommend.feature.common.service.TableStoreService;
+import com.tzld.piaoquan.recommend.feature.produce.service.DTSConfigService;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.spark.api.java.JavaRDD;
+
+import java.util.Map;
+
+@Slf4j
+public class ODPSToTableStore extends AbstractODPSSyncJob {
+
+    private static final TableStoreService tableStoreService = new TableStoreService();
+
+    public static void main(String[] args) {
+        new ODPSToTableStore().run(args);
+    }
+
+    @Override
+    protected String appName(Map<String, String> argMap) {
+        return "ODPSToTableStore : " + argMap.get("table");
+    }
+
+    @Override
+    protected DTSConfig loadConfig(DTSConfigService dtsConfigService, Map<String, String> argMap) {
+        return dtsConfigService.getDTSTableStoreConfig(argMap);
+    }
+
+    @Override
+    protected boolean validate(DTSConfig config) {
+        return config.tableStoreSelfCheck();
+    }
+
+    @Override
+    protected void sink(JavaRDD<Map<String, String>> repartitionedRdd, DTSConfig config, Map<String, String> argMap) {
+        log.info("sync TableStore");
+        repartitionedRdd.foreachPartition(iterator -> tableStoreService.batchWrite(iterator, config));
+    }
+
+}

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

@@ -35,6 +35,10 @@ public class DTSConfigService {
         return this.getDTSConfig(argMap, "dts.config");
     }
 
+    public DTSConfig getDTSTableStoreConfig(Map<String, String> argMap) {
+        return this.getDTSConfig(argMap, "dts.table.store.config");
+    }
+
     private DTSConfig getDTSConfig(Map<String, String> argMap, String apolloConfigKey) {
         List<DTSConfig> dtsConfigs = JSONUtils.fromJson(
                 ConfigService.getAppConfig().getProperty(apolloConfigKey, ""),

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

@@ -6,6 +6,7 @@ import com.tzld.piaoquan.recommend.feature.common.ThreadPoolFactory;
 import com.tzld.piaoquan.recommend.feature.common.model.DTSConfig;
 import com.tzld.piaoquan.recommend.feature.common.util.CommonCollectionUtils;
 import com.tzld.piaoquan.recommend.feature.common.util.CompressionUtil;
+import com.tzld.piaoquan.recommend.feature.common.util.RedisUtil;
 import com.tzld.piaoquan.recommend.feature.model.common.Result;
 import com.tzld.piaoquan.recommend.feature.model.feature.FeatureKeyProto;
 import com.tzld.piaoquan.recommend.feature.model.feature.MultiGetFeatureRequest;
@@ -128,7 +129,7 @@ public class FeatureV2Service {
         return this.buildRedisKey(fk, dtsConfigs);
     }
 
-    // Note：写入和读取的key生成规则应保持一致
+    // Note：写入和读取的key生成规则应保持一致（拼装逻辑统一收敛至 RedisUtil.buildRedisKey）
     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()))
@@ -137,21 +138,6 @@ public class FeatureV2Service {
             log.error("table {} not config", fk.getTableName());
             return "";
         }
-        DTSConfig config = optional.get();
-
-        // Note：写入和读取的key生成规则应保持一致
-        List<String> fields = config.getRedis().getKey();
-        if (org.apache.commons.collections4.CollectionUtils.isEmpty(fields)) {
-            return config.getRedis().getPrefix();
-        }
-        StringBuilder sb = new StringBuilder();
-        if (StringUtils.isNotBlank(config.getRedis().getPrefix())) {
-            sb.append(config.getRedis().getPrefix());
-        }
-        for (String field : fields) {
-            sb.append(":");
-            sb.append(fk.getFieldValueMap().get(field));
-        }
-        return sb.toString();
+        return RedisUtil.buildRedisKey(fk.getFieldValueMap(), optional.get());
     }
 }