public static DataStream<Object> append(
            Configuration conf,
            RowType rowType,
            DataStream<RowData> dataStream,
            boolean bounded,
            MetricOption metricOption) {
        if (!bounded) {
            // In principle, the config should be immutable, but the boundedness
            // is only visible when creating the sink pipeline.
            conf.setBoolean(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT, false);
        }
        WriteOperatorFactory<RowData> operatorFactory = AppendWriteOperator.getFactory(conf, rowType, metricOption);

        return dataStream
                .transform(opName("hoodie_append_write", conf), TypeInformation.of(Object.class), operatorFactory)
                .uid(opUID("hoodie_stream_write", conf))
                .setParallelism(conf.getInteger(FlinkOptions.WRITE_TASKS));
    }
Insert the dataset with append mode(no upsert or deduplication).

<p>The input dataset would be rebalanced among the write tasks:

<pre>
     | input1 | ===\     /=== | task1 | (p1, p2, p3, p4)
                  shuffle
     | input2 | ===/     \=== | task2 | (p1, p2, 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, so there may be many small files.

@param conf       The configuration
@param rowType    The input row type
@param dataStream The input data stream
@param bounded    Whether the input stream is bounded
@return the appending 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