private <T> Builder forMapperOutputType(
                DataStream<T> input, MapFunction<T, RowData> mapper, TypeInformation<RowData> outputType) {
            this.inputCreator =
                    newUidPrefix -> {
                        // Input stream order is crucial for some situation(e.g. in cdc case). Therefore, we
                        // need to set the parallelism
                        // of map operator same as its input to keep map operator chaining its input, and avoid
                        // rebalanced by default.
                        SingleOutputStreamOperator<RowData> inputStream =
                                input.map(mapper, outputType).setParallelism(input.getParallelism());
                        if (newUidPrefix != null) {
                            inputStream.name(operatorName(newUidPrefix)).uid(newUidPrefix + "-mapper");
                        }
                        return inputStream;
                    };
            return this;
        }
Initialize a {@link Builder} to export the data from input data stream with {@link RowData}s
into iceberg table.

@param input the source input data stream with {@link RowData}s.
@return {@link Builder} to connect the iceberg table.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkSink.java