static IcebergProcessOperator<RowData, WriteResult> createStreamWriter(Table table,
            RowType flinkRowType,
            List<Integer> equalityFieldIds,
            FlinkWriteConf flinkWriteConf,
            boolean appendMode,
            String inlongMetric,
            String auditHostAndPorts,
            DirtyOptions dirtyOptions,
            @Nullable DirtySink<Object> dirtySink,
            TableSchema tableSchema,
            boolean switchAppendUpsertEnable,
            boolean miniBatchMode,
            String auditKeys) {
        // flink A, iceberg a
        Preconditions.checkArgument(table != null, "Iceberg table should't be null");

        Table serializableTable = SerializableTable.copyOf(table);
        RowDataTaskWriterFactory taskWriterFactory =
                new RowDataTaskWriterFactory(
                        serializableTable,
                        serializableTable.schema(),
                        flinkRowType,
                        flinkWriteConf.targetDataFileSize(),
                        flinkWriteConf.dataFileFormat(),
                        equalityFieldIds,
                        flinkWriteConf.upsertMode(),
                        appendMode,
                        miniBatchMode);

        RowType tableSchemaRowType = (RowType) tableSchema.toRowDataType().getLogicalType();
        return new IcebergProcessOperator<>(new IcebergSingleStreamWriter<>(
                table.name(), taskWriterFactory, inlongMetric, auditHostAndPorts,
                flinkRowType, dirtyOptions, dirtySink, false,
                tableSchemaRowType, getMetaFieldIndex(tableSchema), switchAppendUpsertEnable, auditKeys));
    }
This operator is used to buffer a mini batch data group by field before writing to save memory resources.

This operator must have the same degree of parallelism as the writer operator to ensure that the data of
the same batch of equality fields in a checkpoint period will only be processed once by a certain writer
subtask (rather than processed multiple times).

@param input
@return
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkSink.java