public static DataStream<Object> hoodieStreamWrite(Configuration conf, DataStream<HoodieRecord> dataStream,
            MetricOption metricOption) {
        if (OptionsResolver.isBucketIndexType(conf)) {
            WriteOperatorFactory<HoodieRecord> operatorFactory =
                    BucketStreamWriteOperator.getFactory(conf, metricOption);
            int bucketNum = conf.getInteger(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS);
            String indexKeyFields = conf.getString(FlinkOptions.INDEX_KEY_FIELD);
            BucketIndexPartitioner<HoodieKey> partitioner = new BucketIndexPartitioner<>(bucketNum, indexKeyFields);
            return dataStream.partitionCustom(partitioner, HoodieRecord::getKey)
                    .transform(opName("bucket_write", conf), TypeInformation.of(Object.class), operatorFactory)
                    .uid(opUID("bucket_write", conf))
                    .setParallelism(conf.getInteger(FlinkOptions.WRITE_TASKS));
        } else {
            WriteOperatorFactory<HoodieRecord> operatorFactory = StreamWriteOperator.getFactory(conf, metricOption);
            return dataStream
                    // Key-by record key, to avoid multiple subtasks write to a bucket at the same time
                    .keyBy(HoodieRecord::getRecordKey)
                    .transform(
                            "bucket_assigner",
                            TypeInformation.of(HoodieRecord.class),
                            new KeyedProcessOperator<>(new BucketAssignFunction<>(conf)))
                    .uid(opUID("bucket_assigner", conf))
                    .setParallelism(conf.getInteger(FlinkOptions.BUCKET_ASSIGN_TASKS))
                    // shuffle by fileId(bucket id)
                    .keyBy(record -> record.getCurrentLocation().getFileId())
                    .transform(opName("stream_write", conf), TypeInformation.of(Object.class), operatorFactory)
                    .uid(opUID("stream_write", conf))
                    .setParallelism(conf.getInteger(FlinkOptions.WRITE_TASKS));
        }
    }
The streaming write pipeline.

<p>The input dataset shuffles by the primary key first then
shuffles by the file group ID before passing around to the write function.
The whole pipeline looks like the following:

<pre>
     | input1 | ===\     /=== | bucket assigner | ===\     /=== | task1 |
                  shuffle(by PK)                    shuffle(by bucket ID)
     | input2 | ===/     \=== | bucket assigner | ===/     \=== | task2 |

     Note: a file group must be handled by one write task to avoid write conflict.
</pre>

<p>The bucket assigner assigns the inputs to suitable file groups, the write task caches
and flushes the data set to disk.

@param conf       The configuration
@param dataStream The input data stream
@return the stream write data stream pipeline
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/hudi/src/main/java/org/apache/inlong/sort/hudi/sink/utils/Pipelines.java