public static DataStreamSink<Object> bulkInsert(Configuration conf, RowType rowType, DataStream<RowData> dataStream,
            MetricOption metricOption) {
        WriteOperatorFactory<RowData> operatorFactory = BulkInsertWriteOperator.getFactory(conf, rowType, metricOption);
        if (OptionsResolver.isBucketIndexType(conf)) {
            String indexKeys = conf.getString(FlinkOptions.INDEX_KEY_FIELD);
            int numBuckets = conf.getInteger(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS);

            BucketIndexPartitioner<HoodieKey> partitioner = new BucketIndexPartitioner<>(numBuckets, indexKeys);
            RowDataKeyGen keyGen = RowDataKeyGen.instance(conf, rowType);
            RowType rowTypeWithFileId = BucketBulkInsertWriterHelper.rowTypeWithFileId(rowType);
            InternalTypeInfo<RowData> typeInfo = InternalTypeInfo.of(rowTypeWithFileId);

            Map<String, String> bucketIdToFileId = new HashMap<>();
            dataStream = dataStream.partitionCustom(partitioner, keyGen::getHoodieKey)
                    .map(record -> BucketBulkInsertWriterHelper.rowWithFileId(bucketIdToFileId, keyGen, record,
                            indexKeys, numBuckets), typeInfo)
                    .setParallelism(conf.getInteger(FlinkOptions.WRITE_TASKS)); // same parallelism as write task to
                                                                                // avoid shuffle
            if (conf.getBoolean(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT)) {
                SortOperatorGen sortOperatorGen = BucketBulkInsertWriterHelper.getFileIdSorterGen(rowTypeWithFileId);
                dataStream = dataStream.transform("file_sorter", typeInfo, sortOperatorGen.createSortOperator())
                        .setParallelism(conf.getInteger(FlinkOptions.WRITE_TASKS)); // same parallelism as write task to
                                                                                    // avoid shuffle
                ExecNodeUtil.setManagedMemoryWeight(dataStream.getTransformation(),
                        conf.getInteger(FlinkOptions.WRITE_SORT_MEMORY) * 1024L * 1024L);
            }
            return dataStream
                    .transform(opName("bucket_bulk_insert", conf), TypeInformation.of(Object.class), operatorFactory)
                    .uid(opUID("bucket_bulk_insert", conf))
                    .setParallelism(conf.getInteger(FlinkOptions.WRITE_TASKS))
                    .addSink(DummySink.INSTANCE)
                    .name("dummy");
        }

        final String[] partitionFields = FilePathUtils.extractPartitionKeys(conf);
        if (partitionFields.length > 0) {
            RowDataKeyGen rowDataKeyGen = RowDataKeyGen.instance(conf, rowType);
            if (conf.getBoolean(FlinkOptions.WRITE_BULK_INSERT_SHUFFLE_INPUT)) {

                // shuffle by partition keys
                // use #partitionCustom instead of #keyBy to avoid duplicate sort operations,
                // see BatchExecutionUtils#applyBatchExecutionSettings for details.
                Partitioner<String> partitioner =
                        (key, channels) -> KeyGroupRangeAssignment.assignKeyToParallelOperator(key,
                                KeyGroupRangeAssignment.computeDefaultMaxParallelism(
                                        conf.getInteger(FlinkOptions.WRITE_TASKS)),
                                channels);
                dataStream = dataStream.partitionCustom(partitioner, rowDataKeyGen::getPartitionPath);
            }
            if (conf.getBoolean(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT)) {
                SortOperatorGen sortOperatorGen = new SortOperatorGen(rowType, partitionFields);
                // sort by partition keys
                dataStream = dataStream
                        .transform("partition_key_sorter",
                                InternalTypeInfo.of(rowType),
                                sortOperatorGen.createSortOperator())
                        .setParallelism(conf.getInteger(FlinkOptions.WRITE_TASKS));
                ExecNodeUtil.setManagedMemoryWeight(dataStream.getTransformation(),
                        conf.getInteger(FlinkOptions.WRITE_SORT_MEMORY) * 1024L * 1024L);
            }
        }
        return dataStream
                .transform(opName("hoodie_bulk_insert_write", conf),
                        TypeInformation.of(Object.class),
                        operatorFactory)
                // follow the parallelism of upstream operators to avoid shuffle
                .setParallelism(conf.getInteger(FlinkOptions.WRITE_TASKS))
                .addSink(DummySink.INSTANCE)
                .name("dummy");
    }
Bulk insert the input dataset at once.

<p>By default, the input dataset would shuffle by the partition path first then
sort by the partition path before passing around to the write function.
The whole pipeline looks like the following:

<pre>
     | input1 | ===\     /=== |sorter| === | task1 | (p1, p2)
                  shuffle
     | input2 | ===/     \=== |sorter| === | task2 | (p3, p4)

     Note: Both input1 and input2's dataset come from partitions: p1, p2, p3, p4
</pre>

<p>The write task switches to new file handle each time it receives a record
from the different partition path, the shuffle and sort would reduce small files.

<p>The bulk insert should be run in batch execution mode.

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