[
https://issues.apache.org/jira/browse/FLINK-28208?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Liu updated FLINK-28208:
------------------------
Description:
The problem is found when using Adaptive Batch Scheduler. In these, a simple
SQL like "select * from * where *" would generate three operators including
source, map and sink. The map's parallelism is set to -1 by default and is not
the same with source and sink. As a result, the three operators can not be
chained together.
The reason is that we add map operator in method createBatchSink but not
setParallelism. The changed code is as following:
{code:java}
private DataStreamSink<Row> createBatchSink(
DataStream<RowData> dataStream,
DataStructureConverter converter,
StorageDescriptor sd,
HiveWriterFactory recordWriterFactory,
OutputFileConfig fileNaming,
final int parallelism)
throws IOException {
...
return dataStream
.map((MapFunction<RowData, Row>) value -> (Row)
converter.toExternal(value))
.setParallelism(parallelism) // New added to ensure the right
parallelism .writeUsingOutputFormat(builder.build())
.setParallelism(parallelism);
} {code}
was:
The problem is found when using Adaptive Batch Scheduler. In these, a simple
SQL like "select * from * where *" would generate three operators including
source, map and sink. The map's parallelism is set to -1 by default and is not
the same with source and sink. As a result, the three operators can not be
chained together.
The reason is that we add map operator in method createBatchSink but not
setParallelism. The changed code is as following:
{code:java}
private DataStreamSink<Row> createBatchSink(
DataStream<RowData> dataStream,
DataStructureConverter converter,
StorageDescriptor sd,
HiveWriterFactory recordWriterFactory,
OutputFileConfig fileNaming,
final int parallelism)
throws IOException {
...
return dataStream
.map((MapFunction<RowData, Row>) value -> (Row)
converter.toExternal(value))
.setParallelism(parallelism) // New added to ensure the right
parallelism .writeUsingOutputFormat(builder.build())
.setParallelism(parallelism);
} {code}
> The method createBatchSink in class HiveTableSink should setParallelism for
> map operator
> ----------------------------------------------------------------------------------------
>
> Key: FLINK-28208
> URL: https://issues.apache.org/jira/browse/FLINK-28208
> Project: Flink
> Issue Type: Improvement
> Components: Connectors / Hive
> Affects Versions: 1.16.0
> Reporter: Liu
> Priority: Major
>
> The problem is found when using Adaptive Batch Scheduler. In these, a simple
> SQL like "select * from * where *" would generate three operators including
> source, map and sink. The map's parallelism is set to -1 by default and is
> not the same with source and sink. As a result, the three operators can not
> be chained together.
> The reason is that we add map operator in method createBatchSink but not
> setParallelism. The changed code is as following:
> {code:java}
> private DataStreamSink<Row> createBatchSink(
> DataStream<RowData> dataStream,
> DataStructureConverter converter,
> StorageDescriptor sd,
> HiveWriterFactory recordWriterFactory,
> OutputFileConfig fileNaming,
> final int parallelism)
> throws IOException {
> ...
> return dataStream
> .map((MapFunction<RowData, Row>) value -> (Row)
> converter.toExternal(value))
> .setParallelism(parallelism) // New added to ensure the right
> parallelism .writeUsingOutputFormat(builder.build())
> .setParallelism(parallelism);
> } {code}
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
