private <T> Builder forMapperOutputType(DataStream<T> input,
                MapFunction<T, RowData> mapper,
                TypeInformation<RowData> outputType) {
            this.inputCreator = newUidPrefix -> {
                if (newUidPrefix != null) {
                    return input.map(mapper, outputType)
                            .name(operatorName(newUidPrefix))
                            .uid(newUidPrefix + "-mapper");
                } else {
                    return input.map(mapper, outputType);
                }
            };
            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.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkSink.java