luoyuxia commented on code in PR #21703:
URL: https://github.com/apache/flink/pull/21703#discussion_r1080813573
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSink.java:
##########
@@ -351,15 +365,211 @@ private DataStreamSink<?> consume(
}
}
- private DataStreamSink<Row> createBatchSink(
+ private DataStreamSink<?> createBatchSink(
+ DataStream<RowData> dataStream,
+ DataStructureConverter converter,
+ HiveWriterFactory recordWriterFactory,
+ TableMetaStoreFactory metaStoreFactory,
+ OutputFileConfig.OutputFileConfigBuilder fileConfigBuilder,
+ String stagingParentDir,
+ StorageDescriptor sd,
+ Properties tableProps,
+ boolean isToLocal,
+ boolean overwrite,
+ int sinkParallelism)
+ throws IOException {
+ org.apache.flink.configuration.Configuration conf =
+ new org.apache.flink.configuration.Configuration();
+ catalogTable.getOptions().forEach(conf::setString);
+ boolean autoCompaction =
conf.getBoolean(FileSystemConnectorOptions.AUTO_COMPACTION);
+ if (autoCompaction) {
+ if (batchShuffleMode != BatchShuffleMode.ALL_EXCHANGES_BLOCKING) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Auto compaction for Hive sink in batch mode
is not supported when the %s is not %s.",
+ ExecutionOptions.BATCH_SHUFFLE_MODE.key(),
+ BatchShuffleMode.ALL_EXCHANGES_BLOCKING));
+ }
+ Optional<Integer> compactParallelismOptional =
+
conf.getOptional(FileSystemConnectorOptions.COMPACTION_PARALLELISM);
+ int compactParallelism =
compactParallelismOptional.orElse(sinkParallelism);
+ return createBatchCompactSink(
+ dataStream,
+ converter,
+ recordWriterFactory,
+ metaStoreFactory,
+ fileConfigBuilder
+ .withPartPrefix(
+ BatchCompactOperator.UNCOMPACTED_PREFIX
+ + "part-"
+ + UUID.randomUUID())
+ .build(),
+ stagingParentDir,
+ sd,
+ tableProps,
+ isToLocal,
+ overwrite,
+ sinkParallelism,
+ compactParallelism);
+ } else {
+ return createBatchNoCompactSink(
+ dataStream,
+ converter,
+ recordWriterFactory,
+ metaStoreFactory,
+ fileConfigBuilder.build(),
+ stagingParentDir,
+ isToLocal,
+ sinkParallelism);
+ }
+ }
+
+ private DataStreamSink<?> createBatchCompactSink(
DataStream<RowData> dataStream,
DataStructureConverter converter,
HiveWriterFactory recordWriterFactory,
TableMetaStoreFactory metaStoreFactory,
OutputFileConfig fileNaming,
String stagingParentDir,
+ StorageDescriptor sd,
+ Properties tableProps,
boolean isToLocal,
- final int parallelism)
+ boolean overwrite,
+ final int sinkParallelism,
+ final int compactParallelism)
+ throws IOException {
+ String[] partitionColumns = getPartitionKeyArray();
+ org.apache.flink.configuration.Configuration conf =
+ new org.apache.flink.configuration.Configuration();
+ catalogTable.getOptions().forEach(conf::setString);
+ HadoopFileSystemFactory fsFactory = fsFactory();
+ org.apache.flink.core.fs.Path tmpPath =
+ new
org.apache.flink.core.fs.Path(toStagingDir(stagingParentDir, jobConf));
+
+ PartitionCommitPolicyFactory partitionCommitPolicyFactory =
+ new PartitionCommitPolicyFactory(
+
conf.get(HiveOptions.SINK_PARTITION_COMMIT_POLICY_KIND),
Review Comment:
No, we don't. In batch mode, it'll always commit partitions even though the
`metastore` policy has been configured.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
