[FLINK-13544][connectors] Set parallelism of table sink operator to i…

…nput transformation parallelism This closes apache#9332
mbalassi · Aug 2, 2019 · 5254126 · 5254126
1 parent cb04628
commit 5254126
Show file tree

Hide file tree

Showing 8 changed files with 17 additions and 3 deletions.
diff --git a/...c/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraAppendTableSink.java b/...c/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraAppendTableSink.java
@@ -92,6 +92,7 @@ public DataStreamSink<?> consumeDataStream(DataStream<Row> dataStream) {
 
  return dataStream
  .addSink(sink)
+ .setParallelism(dataStream.getParallelism())
  .name(TableConnectorUtils.generateRuntimeName(this.getClass(), fieldNames));
 
  }

diff --git a/...org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchUpsertTableSinkBase.java b/...org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchUpsertTableSinkBase.java
@@ -186,6 +186,7 @@ public DataStreamSink<?> consumeDataStream(DataStream<Tuple2<Boolean, Row>> data
  sinkOptions,
  upsertFunction);
  return dataStream.addSink(sinkFunction)
+ .setParallelism(dataStream.getParallelism())
  .name(TableConnectorUtils.generateRuntimeName(this.getClass(), getFieldNames()));
  }
 

diff --git a/...ka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSinkBase.java b/...ka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSinkBase.java
@@ -96,7 +96,10 @@ public DataStreamSink<?> consumeDataStream(DataStream<Row> dataStream) {
  properties,
  serializationSchema,
  partitioner);
- return dataStream.addSink(kafkaProducer).name(TableConnectorUtils.generateRuntimeName(this.getClass(), getFieldNames()));
+ return dataStream
+ .addSink(kafkaProducer)
+ .setParallelism(dataStream.getParallelism())
+ .name(TableConnectorUtils.generateRuntimeName(this.getClass(), getFieldNames()));
  }
 
  @Override

diff --git a/...ctors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCAppendTableSink.java b/...ctors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCAppendTableSink.java
@@ -64,6 +64,7 @@ public static JDBCAppendTableSinkBuilder builder() {
  public DataStreamSink<?> consumeDataStream(DataStream<Row> dataStream) {
  return dataStream
  .addSink(new JDBCSinkFunction(outputFormat))
+ .setParallelism(dataStream.getParallelism())
  .name(TableConnectorUtils.generateRuntimeName(this.getClass(), fieldNames));
  }
 

diff --git a/...ctors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCUpsertTableSink.java b/...ctors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCUpsertTableSink.java
@@ -91,6 +91,7 @@ private JDBCUpsertOutputFormat newFormat() {
  public DataStreamSink<?> consumeDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
  return dataStream
  .addSink(new JDBCUpsertSinkFunction(newFormat()))
+ .setParallelism(dataStream.getParallelism())
  .name(TableConnectorUtils.generateRuntimeName(this.getClass(), schema.getFieldNames()));
  }
 

diff --git a/.../flink-table-api-java-bridge/src/main/java/org/apache/flink/table/sinks/CsvTableSink.java b/.../flink-table-api-java-bridge/src/main/java/org/apache/flink/table/sinks/CsvTableSink.java
@@ -117,6 +117,10 @@ public DataStreamSink<?> consumeDataStream(DataStream<Row> dataStream) {
  if (numFiles > 0) {
  csvRows.setParallelism(numFiles);
  sink.setParallelism(numFiles);
+ } else {
+ // if file number is not set, use input parallelism to make it chained.
+ csvRows.setParallelism(dataStream.getParallelism());
+ sink.setParallelism(dataStream.getParallelism());
  }
 
  sink.name(TableConnectorUtils.generateRuntimeName(CsvTableSink.class, fieldNames));

diff --git a/...ble-api-java-bridge/src/main/java/org/apache/flink/table/sinks/OutputFormatTableSink.java b/...ble-api-java-bridge/src/main/java/org/apache/flink/table/sinks/OutputFormatTableSink.java
@@ -44,6 +44,8 @@ public final void emitDataStream(DataStream<T> dataStream) {
 
  @Override
  public final DataStreamSink<T> consumeDataStream(DataStream<T> dataStream) {
- return dataStream.writeUsingOutputFormat(getOutputFormat());
+ return dataStream
+ .writeUsingOutputFormat(getOutputFormat())
+ .setParallelism(dataStream.getParallelism());
  }
 }
diff --git a/...-common/src/main/java/org/apache/flink/walkthrough/common/table/SpendReportTableSink.java b/...-common/src/main/java/org/apache/flink/walkthrough/common/table/SpendReportTableSink.java
@@ -60,7 +60,8 @@ public void emitDataSet(DataSet<Row> dataSet) {
  public void emitDataStream(DataStream<Row> dataStream) {
  dataStream
  .map(SpendReportTableSink::format)
- .writeUsingOutputFormat(new LoggerOutputFormat());
+ .writeUsingOutputFormat(new LoggerOutputFormat())
+ .setParallelism(dataStream.getParallelism());
  }
 
  @Override