cshuo commented on code in PR #13423: URL: https://github.com/apache/hudi/pull/13423#discussion_r2151163375
########## hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/PipelinesV2.java: ########## @@ -0,0 +1,192 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hudi.sink.utils; + +import org.apache.hudi.client.model.HoodieFlinkInternalRow; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.configuration.OptionsResolver; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.sink.CleanFunction; +import org.apache.hudi.sink.clustering.ClusteringCommitEvent; +import org.apache.hudi.sink.clustering.ClusteringCommitSink; +import org.apache.hudi.sink.clustering.ClusteringOperator; +import org.apache.hudi.sink.clustering.ClusteringPlanEvent; +import org.apache.hudi.sink.clustering.ClusteringPlanOperator; +import org.apache.hudi.sink.compact.CompactOperator; +import org.apache.hudi.sink.compact.CompactionCommitEvent; +import org.apache.hudi.sink.compact.CompactionCommitSink; +import org.apache.hudi.sink.compact.CompactionPlanEvent; +import org.apache.hudi.sink.compact.CompactionPlanOperator; +import org.apache.hudi.sink.v2.HoodieSink; + +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.datastream.DataStreamSink; +import org.apache.flink.streaming.api.operators.ProcessOperator; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil; +import org.apache.flink.table.types.logical.RowType; + +import static org.apache.hudi.sink.utils.Pipelines.opUID; + +/** + * Utilities to generate pipelines for hudi sink V2. + */ +public class PipelinesV2 { + + private static final String SINK_V2_NAME = "sink_v2"; + + public static DataStreamSink<RowData> sink( + DataStream<RowData> dataStream, + Configuration conf, + RowType rowType, + boolean overwrite, + boolean isBounded) { + HoodieSink hoodieSink = new HoodieSink(conf, rowType, overwrite, isBounded); + return dataStream.sinkTo(hoodieSink) + .setParallelism(getParallelismForSinkV2(conf)) + .uid(opUID(SINK_V2_NAME, conf)) + .name(SINK_V2_NAME); + } + + public static DataStream<RowData> composePipeline( + DataStream<RowData> dataStream, + Configuration conf, + RowType rowType, + boolean overwrite, + boolean isBounded) { + // bulk_insert mode + if (OptionsResolver.isBulkInsertOperation(conf)) { + if (!isBounded) { + throw new HoodieException( + "The bulk insert should be run in batch execution mode."); + } + return Pipelines.bulkInsert(conf, rowType, dataStream); + } + + // Append mode + if (OptionsResolver.isAppendMode(conf)) { + // close compaction for append mode + conf.set(FlinkOptions.COMPACTION_SCHEDULE_ENABLED, false); + DataStream<RowData> pipeline = Pipelines.append(conf, rowType, dataStream); + if (OptionsResolver.needsAsyncClustering(conf)) { + return clusterV2(conf, rowType, pipeline); + } else if (OptionsResolver.isLazyFailedWritesCleanPolicy(conf)) { + // add clean function to rollback failed writes for lazy failed writes cleaning policy + return cleanV2(conf, pipeline); + } else { + return pipeline; + } + } + + // process dataStream and write corresponding files + DataStream<RowData> pipeline; + final DataStream<HoodieFlinkInternalRow> hoodieRecordDataStream = Pipelines.bootstrap(conf, rowType, dataStream, isBounded, overwrite); + pipeline = Pipelines.hoodieStreamWrite(conf, rowType, hoodieRecordDataStream); + // compaction + if (OptionsResolver.needsAsyncCompaction(conf)) { + // use synchronous compaction for bounded source. + if (isBounded) { + conf.set(FlinkOptions.COMPACTION_ASYNC_ENABLED, false); + } + return compactV2(conf, pipeline); + } else { + return cleanV2(conf, pipeline); + } + } + + /** + * Get parallelism for Sink V2 writer to make sure the dummy writer operator can be chained with upstream operators. + */ + private static int getParallelismForSinkV2(Configuration conf) { Review Comment: * `getParallelismForSinkV2` is getting parallelism for `HoodieSink#SinkWriter` * the value is parallelism of the final operator of write pipeline which is build in `HoodieSink#addPreWriteTopology`. The write pipeline cannot be pre-built outside `HoodieSink`, so it seems we cannot fetch the parallelism from operator/transformation directly. -- 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]
