[
https://issues.apache.org/jira/browse/FLINK-28208?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17560811#comment-17560811
]
luoyuxia commented on FLINK-28208:
----------------------------------
[~Jiangang] Thanks for reporting it. I think the mapper parallelism should be
set. Would you like to take it since the fix is quite simple?
If you have no time, I'll fix it.
> 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.10#820010)
