[fix] sink add parallelism

FlechazoW · FlechazoW · commit 5b1a82409dc8 · 2020-09-24T16:31:53.000+08:00
diff --git a/cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/CassandraSink.java b/cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/CassandraSink.java
@@ -58,6 +58,8 @@ public class CassandraSink implements RetractStreamTableSink<Row>, IStreamSinkGe
     protected Integer readTimeoutMillis;
     protected Integer connectTimeoutMillis;
     protected Integer poolTimeoutMillis;
+    protected Integer parallelism = -1;
+    protected String registerTableName;
 
     public CassandraSink() {
         // TO DO NOTHING
@@ -78,6 +80,8 @@ public CassandraSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
         this.readTimeoutMillis = cassandraTableInfo.getReadTimeoutMillis();
         this.connectTimeoutMillis = cassandraTableInfo.getConnectTimeoutMillis();
         this.poolTimeoutMillis = cassandraTableInfo.getPoolTimeoutMillis();
+        this.parallelism = cassandraTableInfo.getParallelism();
+        this.registerTableName = cassandraTableInfo.getTableName();
         return this;
     }
 
@@ -106,7 +110,9 @@ public DataStreamSink<Tuple2<Boolean, Row>> consumeDataStream(DataStream<Tuple2<
 
         CassandraOutputFormat outputFormat = builder.finish();
         RichSinkFunction richSinkFunction = new OutputFormatSinkFunction(outputFormat);
-        DataStreamSink dataStreamSink = dataStream.addSink(richSinkFunction);
+        DataStreamSink dataStreamSink = dataStream.addSink(richSinkFunction)
+                .setParallelism(parallelism)
+                .name(registerTableName);
         return dataStreamSink;
     }
 
diff --git a/core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java b/core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java
@@ -34,7 +34,12 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
-import org.apache.calcite.sql.*;
+import org.apache.calcite.sql.SqlBasicCall;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlSelect;
+import org.apache.calcite.sql.SqlWithItem;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
diff --git a/elasticsearch5/elasticsearch5-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/ElasticsearchSink.java b/elasticsearch5/elasticsearch5-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/ElasticsearchSink.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
- 
+
 
 package com.dtstack.flink.sql.sink.elasticsearch;
 
@@ -45,6 +45,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 /**
  * table output elastic5plugin
@@ -77,6 +78,8 @@ public class ElasticsearchSink implements RetractStreamTableSink<Row>, IStreamSi
 
     private int parallelism = -1;
 
+    protected String registerTableName;
+
     private ElasticsearchTableInfo esTableInfo;
 
 
@@ -135,8 +138,6 @@ private RichSinkFunction createEsSinkFunction(){
 
         boolean authMesh = esTableInfo.isAuthMesh();
         if (authMesh) {
-            String username = esTableInfo.getUserName();
-            String password = esTableInfo.getPassword();
             String authPassword = esTableInfo.getUserName() + ":" + esTableInfo.getPassword();
             userConfig.put("xpack.security.user", authPassword);
         }
@@ -154,18 +155,14 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
     @Override
     public DataStreamSink<Tuple2<Boolean, Row>> consumeDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
         RichSinkFunction richSinkFunction = createEsSinkFunction();
-        DataStreamSink streamSink = dataStream.addSink(richSinkFunction);
+        DataStreamSink streamSink = dataStream.addSink(richSinkFunction).name(registerTableName);
         if(parallelism > 0){
             streamSink.setParallelism(parallelism);
         }
 
         return streamSink;
     }
 
-    public void setParallelism(int parallelism) {
-        this.parallelism = parallelism;
-    }
-
     public void setBulkFlushMaxActions(int bulkFlushMaxActions) {
         this.bulkFlushMaxActions = bulkFlushMaxActions;
     }
@@ -183,6 +180,9 @@ public ElasticsearchSink genStreamSink(AbstractTargetTableInfo targetTableInfo)
         String id = elasticsearchTableInfo.getId();
         String[] idField = StringUtils.split(id, ",");
         idIndexList = new ArrayList<>();
+        registerTableName = elasticsearchTableInfo.getName();
+        parallelism = Objects.isNull(elasticsearchTableInfo.getParallelism()) ?
+                parallelism : elasticsearchTableInfo.getParallelism();
 
         for(int i = 0; i < idField.length; ++i) {
             idIndexList.add(Integer.valueOf(idField[i]));
diff --git a/elasticsearch5/elasticsearch5-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/table/ElasticsearchSinkParser.java b/elasticsearch5/elasticsearch5-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/table/ElasticsearchSinkParser.java
@@ -50,6 +50,8 @@ public class ElasticsearchSinkParser extends AbstractTableParser {
 
     private static final String KEY_ES_PASSWORD = "password";
 
+    private static final String KEY_ES_PARALLELISM = "parallelism";
+
     @Override
     protected boolean fieldNameNeedsUpperCase() {
         return false;
@@ -65,6 +67,7 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         elasticsearchTableInfo.setId((String) props.get(KEY_ES_ID_FIELD_INDEX_LIST.toLowerCase()));
         elasticsearchTableInfo.setIndex((String) props.get(KEY_ES_INDEX.toLowerCase()));
         elasticsearchTableInfo.setEsType((String) props.get(KEY_ES_TYPE.toLowerCase()));
+        elasticsearchTableInfo.setParallelism(MathUtil.getIntegerVal(props.getOrDefault(KEY_ES_PARALLELISM.toLowerCase(), 1)));
 
         String authMeshStr = (String)props.get(KEY_ES_AUTHMESH.toLowerCase());
         if (authMeshStr != null & "true".equals(authMeshStr)) {
diff --git a/elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/ElasticsearchSink.java b/elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/ElasticsearchSink.java
@@ -35,9 +35,11 @@
 import com.google.common.collect.Maps;
 import org.apache.commons.lang.StringUtils;
 import org.apache.http.HttpHost;
+
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.stream.Collectors;
 
 /**
@@ -69,6 +71,8 @@ public class ElasticsearchSink implements RetractStreamTableSink<Row>, IStreamSi
 
     private int parallelism = -1;
 
+    protected String registerTableName;
+
     private ElasticsearchTableInfo esTableInfo;
 
 
@@ -142,6 +146,8 @@ public ElasticsearchSink genStreamSink(AbstractTargetTableInfo targetTableInfo)
         columnTypes = esTableInfo.getFieldTypes();
         esAddressList = Arrays.asList(esTableInfo.getAddress().split(","));
         String id = esTableInfo.getId();
+        registerTableName = esTableInfo.getName();
+        parallelism = Objects.isNull(esTableInfo.getParallelism()) ? parallelism : esTableInfo.getParallelism();
 
         if (!StringUtils.isEmpty(id)) {
             idIndexList = Arrays.stream(StringUtils.split(id, ",")).map(Integer::valueOf).collect(Collectors.toList());
diff --git a/elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/table/ElasticsearchSinkParser.java b/elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/table/ElasticsearchSinkParser.java
@@ -49,6 +49,8 @@ public class ElasticsearchSinkParser extends AbstractTableParser {
 
     private static final String KEY_TRUE = "true";
 
+    private static final String KEY_ES6_PARALLELISM = "parallelism";
+
     @Override
     protected boolean fieldNameNeedsUpperCase() {
         return false;
@@ -64,6 +66,7 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         elasticsearchTableInfo.setId((String) props.get(KEY_ES6_ID_FIELD_INDEX_LIST.toLowerCase()));
         elasticsearchTableInfo.setIndex((String) props.get(KEY_ES6_INDEX.toLowerCase()));
         elasticsearchTableInfo.setEsType((String) props.get(KEY_ES6_TYPE.toLowerCase()));
+        elasticsearchTableInfo.setParallelism(MathUtil.getIntegerVal(props.getOrDefault(KEY_ES6_PARALLELISM.toLowerCase(), 1)));
 
         String authMeshStr = (String) props.get(KEY_ES6_AUTHMESH.toLowerCase());
         if (authMeshStr != null && StringUtils.equalsIgnoreCase(KEY_TRUE, authMeshStr)) {
diff --git a/hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseSink.java b/hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseSink.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
- 
+
 
 package com.dtstack.flink.sql.sink.hbase;
 
@@ -121,7 +121,12 @@ public DataStreamSink<Tuple2<Boolean, Row>> consumeDataStream(DataStream<Tuple2<
 
         HbaseOutputFormat outputFormat = builder.finish();
         RichSinkFunction richSinkFunction = new OutputFormatSinkFunction(outputFormat);
-        DataStreamSink dataStreamSink = dataStream.addSink(richSinkFunction);
+        DataStreamSink dataStreamSink = dataStream.addSink(richSinkFunction).name(registerTabName);
+
+        if (parallelism > 0) {
+            dataStreamSink.setParallelism(parallelism);
+        }
+        
         return dataStreamSink;
     }
 
diff --git a/kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/AbstractKafkaSink.java b/kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/AbstractKafkaSink.java
@@ -58,7 +58,7 @@ public abstract class AbstractKafkaSink implements RetractStreamTableSink<Row>,
     protected String[] partitionKeys;
     protected String sinkOperatorName;
     protected Properties properties;
-    protected int parallelism;
+    protected int parallelism = -1;
     protected String topic;
     protected String tableName;
     protected String updateMode;
diff --git a/kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduSink.java b/kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduSink.java
@@ -56,6 +56,7 @@ public KuduSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
         this.keytab = kuduTableInfo.getKeytab();
         this.krb5conf = kuduTableInfo.getKrb5conf();
         this.enableKrb = kuduTableInfo.isEnableKrb();
+        this.parallelism = kuduTableInfo.getParallelism();
 
         return this;
     }
diff --git a/mongo/mongo-sink/src/main/java/com/dtstack/flink/sql/sink/mongo/MongoSink.java b/mongo/mongo-sink/src/main/java/com/dtstack/flink/sql/sink/mongo/MongoSink.java
@@ -34,6 +34,8 @@
 import org.apache.flink.table.sinks.TableSink;
 import org.apache.flink.types.Row;
 
+import java.util.Objects;
+
 /**
  * Reason:
  * Date: 2018/11/6
@@ -49,6 +51,8 @@ public class MongoSink implements RetractStreamTableSink<Row>, IStreamSinkGener<
     protected String userName;
     protected String password;
     protected String database;
+    protected Integer parallelism = 1;
+    protected String registerTableName;
 
     public MongoSink() {
         // TO DO NOTHING
@@ -62,6 +66,9 @@ public MongoSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
         this.userName = mongoTableInfo.getUserName();
         this.password = mongoTableInfo.getPassword();
         this.database = mongoTableInfo.getDatabase();
+        this.parallelism = Objects.isNull(mongoTableInfo.getParallelism()) ?
+                parallelism : mongoTableInfo.getParallelism();
+        this.registerTableName = mongoTableInfo.getName();
         return this;
     }
 
@@ -83,7 +90,9 @@ public DataStreamSink<Tuple2<Boolean, Row>> consumeDataStream(DataStream<Tuple2<
 
         MongoOutputFormat outputFormat = builder.finish();
         RichSinkFunction richSinkFunction = new OutputFormatSinkFunction(outputFormat);
-        DataStreamSink dataStreamSink = dataStream.addSink(richSinkFunction);
+        DataStreamSink dataStreamSink = dataStream.addSink(richSinkFunction)
+                .setParallelism(parallelism)
+                .name(registerTableName);
         return dataStreamSink;
     }
 
diff --git a/redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisSink.java b/redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisSink.java
@@ -34,6 +34,8 @@
 import org.apache.flink.types.Row;
 
 import java.util.List;
+import java.util.Objects;
+
 /**
  * @author yanxi
  */
@@ -65,6 +67,10 @@ public class RedisSink implements RetractStreamTableSink<Row>, IStreamSinkGener<
 
     protected String masterName;
 
+    protected Integer parallelism = 1;
+
+    protected String registerTableName;
+
     public RedisSink(){
 
     }
@@ -83,6 +89,9 @@ public RedisSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
         this.minIdle = redisTableInfo.getMinIdle();
         this.masterName = redisTableInfo.getMasterName();
         this.timeout = redisTableInfo.getTimeout();
+        this.parallelism = Objects.isNull(redisTableInfo.getParallelism()) ?
+                parallelism : redisTableInfo.getParallelism();
+        this.registerTableName = redisTableInfo.getName();
         return this;
     }
 
@@ -114,7 +123,9 @@ public DataStreamSink<Tuple2<Boolean, Row>> consumeDataStream(DataStream<Tuple2<
                 .setMasterName(this.masterName);
         RedisOutputFormat redisOutputFormat = builder.finish();
         RichSinkFunction richSinkFunction = new OutputFormatSinkFunction(redisOutputFormat);
-        DataStreamSink dataStreamSink = dataStream.addSink(richSinkFunction);
+        DataStreamSink dataStreamSink = dataStream.addSink(richSinkFunction)
+                .setParallelism(parallelism)
+                .name(registerTableName);
         return dataStreamSink;
     }
 
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
@@ -55,6 +55,7 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
             primaryKeysList = Arrays.asList(StringUtils.split(primaryKeysStr, ","));
         }
         redisTableInfo.setPrimaryKeys(primaryKeysList);
+        redisTableInfo.setParallelism(MathUtil.getIntegerVal(props.get(RedisTableInfo.PARALLELISM_KEY.toLowerCase())));
 
         return redisTableInfo;
     }

Original file line number	Diff line number	Diff line change
`@@ -56,6 +56,7 @@ public KuduSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {`
`56`	`56`	`this.keytab = kuduTableInfo.getKeytab();`
`57`	`57`	`this.krb5conf = kuduTableInfo.getKrb5conf();`
`58`	`58`	`this.enableKrb = kuduTableInfo.isEnableKrb();`
	`59`	`+ this.parallelism = kuduTableInfo.getParallelism();`
`59`	`60`
`60`	`61`	`return this;`
`61`	`62`	`}`