feat:add delete redis job

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

@@ -0,0 +1,83 @@
+package com.tzld.piaoquan.recommend.feature.produce;
+
+import com.tzld.piaoquan.recommend.feature.produce.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.produce.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.getV2DTSConfig(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/service/RedisService.java

@@ -77,6 +77,26 @@ public class RedisService implements Serializable {
         }
     }
 
+    public void delete(Iterator<Map<String, String>> dataIte, DTSConfig config) {
+        log.info("delete param: {}", config);
+        List<String> batch = new ArrayList<>();
+        Jedis jedis = new Jedis(hostName, port);
+        jedis.auth(password);
+        while (dataIte.hasNext()) {
+            Map<String, String> record = dataIte.next();
+            String redisKey = buildRedisKey(record, config);
+            batch.add(redisKey);
+            if (batch.size() >= 1000) {
+                delete(jedis, batch);
+                batch.clear();
+            }
+        }
+        if (CollectionUtils.isNotEmpty(batch)) {
+            delete(jedis, batch);
+        }
+        jedis.close();
+    }
+
     private void mSet(Jedis jedis, Map<String, String> batch, long expire, TimeUnit timeUnit) {
         long expireSeconds = timeUnit.toSeconds(expire);
 
@@ -92,7 +112,19 @@ public class RedisService implements Serializable {
 
     }
 
-    public String buildRedisKey(Map<String, String> fieldValueMap, DTSConfig config) {
+    private void delete(Jedis jedis, List<String> keys) {
+        Pipeline pipeline = jedis.pipelined();
+        for (String key : keys) {
+            try {
+                pipeline.del(String.format("%s:v2", key).getBytes(StandardCharsets.UTF_8));
+            } catch (Exception ignore) {
+
+            }
+        }
+        pipeline.sync();
+    }
+
+    private String buildRedisKey(Map<String, String> fieldValueMap, DTSConfig config) {
         // Note：写入和读取的key生成规则应保持一致
         List<String> fields = config.getRedis().getKey();
         if (CollectionUtils.isEmpty(fields)) {
@@ -109,7 +141,7 @@ public class RedisService implements Serializable {
         return sb.toString();
     }
 
-    public Map<String, String> buildValueMap(Map<String, String> record, DTSConfig config) {
+    private Map<String, String> buildValueMap(Map<String, String> record, DTSConfig config) {
         Map<String, String> valueMap = new HashMap<>();
         if (StringUtils.isNotBlank(config.getRedis().getFeatureField())) {
             valueMap = JSONUtils.fromJson(record.get(config.getRedis().getFeatureField()), new TypeToken<Map<String, String>>() {