Merge remote-tracking branch 'origin/v1.5.0_dev' into v1.8.0_dev

simenliuxing · simenliuxing · commit faea88dabffa · 2019-11-11T14:44:56.000+08:00
# Conflicts:
#	redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java
diff --git a/docs/redisSide.md b/docs/redisSide.md
@@ -11,6 +11,7 @@
      password = 'redisPwd',
      database = 'dbName',
      tableName ='sideTableName',
+     redisType = '1',
      cache ='LRU',
      cacheSize ='10000',
      cacheTTLMs ='60000'
@@ -35,6 +36,8 @@
 | type | 表明维表的类型[hbase\|mysql\|redis]|是||
 | url | redis 的地址;格式ip:port[,ip:port]|是||
 | password | redis 的密码 |是||
+| redisType | redis模式（1 单机，2 哨兵， 3 集群）| 是 |
+| masterName | 主节点名称（哨兵模式下为必填项） | 否 |
 | database | reids 的数据库地址|否||
 | tableName | redis 的表名称|是||
 | cache | 维表缓存策略(NONE/LRU/ALL)|否|NONE|
@@ -60,6 +63,7 @@ create table sideTable(
     url='172.16.10.79:6379',
     password='abc123',
     database='0',
+    redisType = '1',
     tableName='sidetest',
     cache = 'LRU',
     cacheTTLMs='10000'
diff --git a/docs/redisSink.md b/docs/redisSink.md
@@ -9,6 +9,7 @@ CREATE TABLE tableName(
     url = 'ip:port',
     database ='dbName',
     password ='pwd',
+    redisType='1',
     tableName ='tableName',
     parallelism ='parllNum'
  );
@@ -32,12 +33,14 @@ redis5.0
   
 |参数名称|含义|是否必填|默认值|
 |----|---|---|-----|
-|type | 表明 输出表类型[mysql\|hbase\|elasticsearch\|redis\]|是||
+| type | 表明 输出表类型[mysql\|hbase\|elasticsearch\|redis\]|是||
 | url | redis 的地址;格式ip:port[,ip:port]|是||
 | password | redis 的密码 |是||
+| redisType | redis模式（1 单机，2 哨兵， 3 集群）| 是 |
+| masterName | 主节点名称（哨兵模式下为必填项） | 否 |
 | database | reids 的数据库地址|否||
 | tableName | redis 的表名称|是||
-|parallelism | 并行度设置|否|1|
+| parallelism | 并行度设置|否|1|
       
   
 ## 5.样例：
@@ -51,7 +54,8 @@ redis5.0
     url='172.16.10.79:6379',
     password='abc123',
     database='0',
-    tableName='sinktoredis',
+    redisType='1',
+    tableName='sinktoredis'
  );
 
  ```
diff --git a/redis5/redis5-side/redis-all-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAllReqRow.java b/redis5/redis5-side/redis-all-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAllReqRow.java
@@ -98,7 +98,7 @@ public void flatMap(Row row, Collector<Row> out) throws Exception {
                 out.collect(null);
             }
             String columnName = sideInfo.getEqualFieldList().get(conValIndex);
-            inputParams.put(columnName, (String) equalObj);
+            inputParams.put(columnName, equalObj.toString());
         }
         String key = buildKey(inputParams);
 
diff --git a/redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java b/redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java
@@ -36,10 +36,8 @@
 import org.apache.flink.calcite.shaded.com.google.common.collect.Maps;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
-import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
 import org.apache.flink.types.Row;
 
-import java.sql.Timestamp;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -124,12 +122,12 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
             Integer conValIndex = sideInfo.getEqualValIndex().get(i);
             Object equalObj = input.getField(conValIndex);
             if(equalObj == null){
-                resultFuture.complete(null);
+                dealMissKey(input, resultFuture);
                 return;
             }
-
+            String value = equalObj.toString();
             keyData.add(sideInfo.getEqualFieldList().get(i));
-            keyData.add((String) equalObj);
+            keyData.add(value);
         }
 
         String key = buildCacheKey(keyData);
@@ -159,29 +157,33 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
         Map<String, String> keyValue = Maps.newHashMap();
         List<String> value = async.keys(key + ":*").get();
         String[] values = value.toArray(new String[value.size()]);
-        RedisFuture<List<KeyValue<String, String>>> future =  ((RedisStringAsyncCommands) async).mget(values);
-        future.thenAccept(new Consumer<List<KeyValue<String, String>>>() {
-            @Override
-            public void accept(List<KeyValue<String, String>> keyValues) {
-                if (keyValues.size() != 0){
-                    for (int i=0; i<keyValues.size(); i++){
-                        String[] splitKeys = keyValues.get(i).getKey().split(":");
-                        keyValue.put(splitKeys[1], splitKeys[2]);
-                        keyValue.put(splitKeys[3], keyValues.get(i).getValue());
-                    }
-                    Row row = fillData(input, keyValue);
-                    resultFuture.complete(Collections.singleton(row));
-                    if(openCache()){
-                        putCache(key, CacheObj.buildCacheObj(ECacheContentType.MultiLine, keyValue));
-                    }
-                } else {
-                    dealMissKey(input, resultFuture);
-                    if(openCache()){
-                        putCache(key, CacheMissVal.getMissKeyObj());
+        if (values.length == 0){
+            dealMissKey(input, resultFuture);
+        } else {
+            RedisFuture<List<KeyValue<String, String>>> future = ((RedisStringAsyncCommands) async).mget(values);
+            future.thenAccept(new Consumer<List<KeyValue<String, String>>>() {
+                @Override
+                public void accept(List<KeyValue<String, String>> keyValues) {
+                    if (keyValues.size() != 0) {
+                        for (int i = 0; i < keyValues.size(); i++) {
+                            String[] splitKeys = keyValues.get(i).getKey().split(":");
+                            keyValue.put(splitKeys[1], splitKeys[2]);
+                            keyValue.put(splitKeys[3], keyValues.get(i).getValue());
+                        }
+                        Row row = fillData(input, keyValue);
+                        resultFuture.complete(Collections.singleton(row));
+                        if (openCache()) {
+                            putCache(key, CacheObj.buildCacheObj(ECacheContentType.MultiLine, keyValue));
+                        }
+                    } else {
+                        dealMissKey(input, resultFuture);
+                        if (openCache()) {
+                            putCache(key, CacheMissVal.getMissKeyObj());
+                        }
                     }
                 }
-            }
-        });
+            });
+        }
     }
 
     private String buildCacheKey(List<String> keyData) {
diff --git a/redis5/redis5-side/redis-side-core/src/main/java/com/dtstack/flink/sql/side/redis/table/RedisSideReqRow.java b/redis5/redis5-side/redis-side-core/src/main/java/com/dtstack/flink/sql/side/redis/table/RedisSideReqRow.java
@@ -24,6 +24,8 @@
 import org.apache.flink.types.Row;
 
 import java.io.Serializable;
+import java.math.BigDecimal;
+import java.sql.Date;
 import java.sql.Timestamp;
 import java.util.Map;
 
@@ -63,10 +65,50 @@ public Row fillData(Row input, Object sideInput) {
                 row.setField(entry.getKey(), null);
             }else{
                 String key = sideInfo.getSideFieldNameIndex().get(entry.getKey());
-                row.setField(entry.getKey(), sideInputMap.get(key));
+                setRowField(row, entry.getKey(), sideInfo, sideInputMap.get(key));
             }
         }
 
         return row;
     }
+
+    public void setRowField(Row row, Integer index, SideInfo sideInfo, String value) {
+        Integer keyIndex = sideInfo.getSideFieldIndex().get(index);
+        String classType = sideInfo.getSideTableInfo().getFieldClassList().get(keyIndex).getName();
+        switch (classType){
+            case "java.lang.Integer":
+                row.setField(index, Integer.valueOf(value));
+                break;
+            case "java.lang.String":
+                row.setField(index, value);
+                break;
+            case "java.lang.Double":
+                row.setField(index, Double.valueOf(value));
+                break;
+            case "java.lang.Long":
+                row.setField(index, Long.valueOf(value));
+                break;
+            case "java.lang.Byte":
+                row.setField(index, Byte.valueOf(value));
+                break;
+            case "java.lang.Short":
+                row.setField(index, Short.valueOf(value));
+                break;
+            case "java.lang.Float":
+                row.setField(index, Float.valueOf(value));
+                break;
+            case "java.math.BigDecimal":
+                row.setField(index, BigDecimal.valueOf(Long.valueOf(value)));
+                break;
+            case "java.sql.Timestamp":
+                row.setField(index, Timestamp.valueOf(value));
+                break;
+            case "java.sql.Date":
+                row.setField(index, Date.valueOf(value));
+                break;
+            default:
+                throw new RuntimeException("no support field type. the type: " + classType);
+        }
+    }
+
 }
diff --git a/redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisOutputFormat.java b/redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisOutputFormat.java
@@ -171,7 +171,13 @@ public void writeRecord(Tuple2 record) throws IOException {
         for (int i = 0; i < fieldNames.length; i++) {
             StringBuilder key = new StringBuilder();
             key.append(tableName).append(":").append(perKey).append(":").append(fieldNames[i]);
-            jedis.set(key.toString(), row.getField(i).toString());
+
+            String value = "null";
+            Object field = row.getField(i);
+            if (field != null) {
+                value = field.toString();
+            }
+            jedis.set(key.toString(), value);
         }
 
         if (outRecords.getCount()%rowLenth == 0){
diff --git a/redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/table/RedisSinkParser.java b/redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/table/RedisSinkParser.java
@@ -21,7 +21,10 @@
 import com.dtstack.flink.sql.table.AbsTableParser;
 import com.dtstack.flink.sql.table.TableInfo;
 import com.dtstack.flink.sql.util.MathUtil;
+import org.apache.commons.lang3.StringUtils;
 
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Map;
 
 public class RedisSinkParser extends AbsTableParser {
@@ -42,6 +45,17 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
         redisTableInfo.setMinIdle(MathUtil.getString(props.get(RedisTableInfo.MINIDLE.toLowerCase())));
         redisTableInfo.setRedisType(MathUtil.getString(props.get(RedisTableInfo.REDIS_TYPE.toLowerCase())));
         redisTableInfo.setMasterName(MathUtil.getString(props.get(RedisTableInfo.MASTER_NAME.toLowerCase())));
+
+        String primaryKeysStr = MathUtil.getString(props.get("primarykeys"));
+        ArrayList<String> primaryKeysList = null;
+        if (!StringUtils.isEmpty(primaryKeysStr)) {
+            String[] primaryKeysArray = primaryKeysStr.split(",");
+            primaryKeysList = new ArrayList<String>(Arrays.asList(primaryKeysArray));
+        } else {
+            primaryKeysList = new ArrayList<>();
+        }
+        redisTableInfo.setPrimaryKeys(primaryKeysList);
+
         return redisTableInfo;
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -98,7 +98,7 @@ public void flatMap(Row row, Collector<Row> out) throws Exception {`
`98`	`98`	`out.collect(null);`
`99`	`99`	`}`
`100`	`100`	`String columnName = sideInfo.getEqualFieldList().get(conValIndex);`
`101`		`- inputParams.put(columnName, (String) equalObj);`
	`101`	`+ inputParams.put(columnName, equalObj.toString());`
`102`	`102`	`}`
`103`	`103`	`String key = buildKey(inputParams);`
`104`	`104`