Merge remote-tracking branch 'origin/feat_1.8_redisSink' into hotfix_1.8_3.10.x_27917

Author: dapeng

docs/plugin/redisSink.md

@@ -94,7 +94,8 @@ redis5.0
 
 ## 6.redis完整样例
 ### redis数据说明
-redis使用k-v格式存储，key的构建格式为tableName:privateKey:privateKeyValue:columnName, value=columnValue
+redis使用散列类型 hash 数据结构，key=tableName_primaryKey1_primaryKey2,value={column1=value1, column2=value2}
+如果以班级class表为例，id和name作为联合主键，那么redis的结构为 <class_1_john ,{id=1, name=john, age=12}>
 
 ### 源表数据内容
 ```
@@ -103,10 +104,10 @@ redis使用k-v格式存储，key的构建格式为tableName:privateKey:privateKe
 ### redis实际数据内容
 ```
 127.0.0.1:6379> keys *
-1) "resultTable:name:roc:name"
-2) "resultTable:name:roc:channel"
-127.0.0.1:6379> get "resultTable:name:roc:name"
-"roc"
-127.0.0.1:6379> get "resultTable:name:roc:channel"
-"daishu"
+1) "resultTable_roc"
+127.0.0.1:6379> hgetall resultTable_roc
+1) "channel"
+2) "daishu"
+3) "name"
+4) "roc"
 ```

redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisOutputFormat.java

@@ -19,6 +19,8 @@
 package com.dtstack.flink.sql.sink.redis;
 
 import com.dtstack.flink.sql.outputformat.AbstractDtRichOutputFormat;
+import com.dtstack.flink.sql.sink.redis.enums.RedisType;
+import com.google.common.collect.Maps;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
@@ -35,11 +37,7 @@
 
 import java.io.Closeable;
 import java.io.IOException;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Set;
+import java.util.*;
 
 /**
  * @author yanxi
@@ -49,7 +47,7 @@ public class RedisOutputFormat extends AbstractDtRichOutputFormat<Tuple2> {
 
     private String url;
 
-    private String database;
+    private String database = "0";
 
     private String tableName;
 
@@ -71,7 +69,7 @@ public class RedisOutputFormat extends AbstractDtRichOutputFormat<Tuple2> {
 
     protected List<String> primaryKeys;
 
-    protected int timeout;
+    protected int timeout = 10000;
 
     private JedisPool pool;
 
@@ -121,29 +119,21 @@ private void establishConnection() {
             String[] ipPortPair = StringUtils.split(ipPort, ":");
             addresses.add(new HostAndPort(ipPortPair[0].trim(), Integer.valueOf(ipPortPair[1].trim())));
         }
-        if (timeout == 0){
-            timeout = 10000;
-        }
-        if (database == null)
-        {
-            database = "0";
-        }
 
-        switch (redisType){
-            //单机
-            case 1:
+        switch (RedisType.parse(redisType)){
+            case STANDALONE:
                 pool = new JedisPool(poolConfig, firstIp, Integer.parseInt(firstPort), timeout, password, Integer.parseInt(database));
                 jedis = pool.getResource();
                 break;
-            //哨兵
-            case 2:
+            case SENTINEL:
                 jedisSentinelPool = new JedisSentinelPool(masterName, ipPorts, poolConfig, timeout, password, Integer.parseInt(database));
                 jedis = jedisSentinelPool.getResource();
                 break;
-            //集群
-            case 3:
+            case CLUSTER:
                 jedis = new JedisCluster(addresses, timeout, timeout, 10, password, poolConfig);
+                break;
             default:
+                throw new RuntimeException("unsupport redis type[ " + redisType + "]");
         }
     }
 
@@ -158,36 +148,14 @@ public void writeRecord(Tuple2 record) throws IOException {
         if (row.getArity() != fieldNames.length) {
             return;
         }
-
-        HashMap<String, Integer> map = new HashMap<>(8);
-        for (String primaryKey : primaryKeys) {
-            for (int i = 0; i < fieldNames.length; i++) {
-                if (fieldNames[i].equals(primaryKey)) {
-                    map.put(primaryKey, i);
-                }
-            }
-        }
-
-        List<String> kvList = new LinkedList<>();
-        for (String primaryKey : primaryKeys){
-            StringBuilder primaryKv = new StringBuilder();
-            int index = map.get(primaryKey).intValue();
-            primaryKv.append(primaryKey).append(":").append(row.getField(index));
-            kvList.add(primaryKv.toString());
-        }
-
-        String perKey = String.join(":", kvList);
-        for (int i = 0; i < fieldNames.length; i++) {
-            StringBuilder key = new StringBuilder();
-            key.append(tableName).append(":").append(perKey).append(":").append(fieldNames[i]);
-
-            String value = "null";
-            Object field = row.getField(i);
-            if (field != null) {
-                value = field.toString();
-            }
-            jedis.set(key.toString(), value);
+        Map<String, Object> refData = Maps.newHashMap();
+        for(int i = 0; i < fieldNames.length; i++){
+            refData.put(fieldNames[i], row.getField(i));
         }
+        String redisKey = buildCacheKey(refData);
+        refData.entrySet().forEach(e ->{
+            jedis.hset(redisKey, e.getKey(), String.valueOf(e.getValue()));
+        });
 
         if (outRecords.getCount() % ROW_PRINT_FREQUENCY == 0){
             LOG.info(record.toString());
@@ -211,6 +179,17 @@ public void close() throws IOException {
 
     }
 
+    public String buildCacheKey(Map<String, Object> refData) {
+        StringBuilder keyBuilder = new StringBuilder(tableName);
+        for(String primaryKey : primaryKeys){
+            if(!refData.containsKey(primaryKey)){
+                return null;
+            }
+            keyBuilder.append("_").append(refData.get(primaryKey));
+        }
+        return keyBuilder.toString();
+    }
+
     public static RedisOutputFormatBuilder buildRedisOutputFormat(){
         return new RedisOutputFormatBuilder();
     }

redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/enums/RedisType.java

@@ -0,0 +1,33 @@
+package com.dtstack.flink.sql.sink.redis.enums;
+
+public enum RedisType {
+    /**
+     * 单机
+     */
+    STANDALONE(1),
+    /**
+     * 哨兵
+     */
+    SENTINEL(2),
+    /**
+     * 集群
+     */
+    CLUSTER(3);
+    int type;
+    RedisType(int type){
+        this.type = type;
+    }
+
+    public int getType(){
+        return type;
+    }
+
+    public static RedisType parse(int redisType){
+        for(RedisType type : RedisType.values()){
+            if(type.getType() == redisType){
+                return type;
+            }
+        }
+        throw new RuntimeException("unsupport redis type["+ redisType + "]");
+    }
+}