You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "luoyuxia (Jira)" <ji...@apache.org> on 2022/06/30 04:08:00 UTC
[jira] [Commented] (FLINK-28208) The method createBatchSink in class HiveTableSink should setParallelism for map operator
[ 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)