kafka11 parallelism

todd5167 · todd5167 · commit 72ba34eab371 · 2019-09-24T12:16:58.000+08:00
diff --git a/kafka11/kafka11-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java b/kafka11/kafka11-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java
@@ -98,7 +98,7 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
 
         DataStream<Row> ds = dataStream.map((Tuple2<Boolean, Row> record) -> {
             return record.f1;
-        }).returns(getOutputType().getTypeAt(1));
+        }).returns(getOutputType().getTypeAt(1)).setParallelism(parallelism);
 
         kafkaTableSink.emitDataStream(ds);
     }

Original file line number	Diff line number	Diff line change
`@@ -98,7 +98,7 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {`
`98`	`98`
`99`	`99`	`DataStream<Row> ds = dataStream.map((Tuple2<Boolean, Row> record) -> {`
`100`	`100`	`return record.f1;`
`101`		`- }).returns(getOutputType().getTypeAt(1));`
	`101`	`+ }).returns(getOutputType().getTypeAt(1)).setParallelism(parallelism);`
`102`	`102`
`103`	`103`	`kafkaTableSink.emitDataStream(ds);`
`104`	`104`	`}`