Flink: change sink shuffle to use RowData as data type and statistics key type

[stevenzwu]

, because FlinkSink normalize the data type to RowData before coming to writer. Also added custom type serializers for MapDataStatistics and /DataStatisticsOrRecord.

[stevenzwu]

address some of the comments from issue #7393.

@huyuanfeng2018 @hililiwei @yegangy0718 can you help review?

@yegangy0718 will follow up with a separate PR on jmh benchmark.

[stevenzwu]

Factory is replaced by TypeSerializer

[stevenzwu]

used generic trick for the strong type check. it shouldn't matter to users since all these are internal classes.

[stevenzwu]

use the base interface DataStatistics so that it can be used by DataStatisticsOrRecordSerializer.

DataStatisticsOrRecordSerializer(
      TypeSerializer<DataStatistics<D, S>> statisticsSerializer,
      TypeSerializer<RowData> recordSerializer)

[stevenzwu]

need to wrap the block in try with test harness. otherwise, Flink don't know how to serialize the output type ofDataStatisticsOrRecord <>. test harness has the proper setup(...) on output type serializer

[stevenzwu]

we will use RowDataProjection to extract the key. PR #7493 is related.

[huyuanfeng2018]

Is it necessary to add a method here to count statistical information with values? For example, data under different keys may have different sizes. Is it also a consideration when controlling subsequent balance，like： add(Rowdata key, V v) Among them, v may represent the record bytes of the row corresponding to the current key. What do you think？

[stevenzwu]

we are only counting records per key. To get the bytes, it would require serialization or some other trick of estimation. Agree bytes would be the best. but record count is probably also good enough.

[huyuanfeng2018]

Ok, I think this DataStatisticsOperator is good for collecting some statistical information. It seems that we also need an Operator to determine the partitionID, such as PartitionIdAssignerOperator and then pass org.apache.flink.api.java.functions.IdPartitioner From custom data distribution, I haven’t seen the design of this piece in the design document, can you briefly introduce the follow-up implementation

[stevenzwu]

Yes, this is only for collecting statistics to guide the partitioner decision. we will implement a custom range partitioner (Flink ) that splits the value into ranges (one for each assigner writer subtask) based on the statistics.

[hililiwei]

According to my understanding, each writer will be responsible for one or more partitions, and we will distribute the data arriving from upstream to the corresponding writers, right?

[stevenzwu]

@hililiwei that is correct. custom range partitioner for Flink DataStream will distribute the data to writer subtasks with good clustering based on the data statistics.

[pvary]

@stevenzwu: I like that with a correct reader we can make sure that the IcebergSource could read whatever data structure we want. It would be good to have the possibility for the same thing for the FlinkSink (or IcebergSink later). Wouldn't this PR close this possibility by forcing the Sink to use RowData?

[stevenzwu]

@stevenzwu: I like that with a correct reader we can make sure that the IcebergSource could read whatever data structure we want. It would be good to have the possibility for the same thing for the FlinkSink (or IcebergSink later). Wouldn't this PR close this possibility by forcing the Sink to use RowData?

@pvary FlinkSink can still support other input types (like Avro GenericRecord). The behavior (internally) is that all input type is converted to Flink RowData, which is the only supported record format for the underline file writer in Iceberg.

That is the reason why @huyuanfeng2018 was suggesting if RowData is the only key type we need to support. If in the future, Flink sink supports other record format for the underline file writer, maybe we can revisit this decision. for now, RowData is the only type Flink file writer supports.

[hililiwei]

We only care about the amount of data, not the size of data, right?

[stevenzwu]

@hililiwei not sure if I understand the question here. can you elaborate?

here is the javadoc for this method from the TypeSerializer interface.

Returns:
The length of the data type, or -1 for variable length data types.

[pvary]

@pvary FlinkSink can still support other input types (like Avro GenericRecord). The behavior (internally) is that all input type is converted to Flink RowData, which is the only supported record format for the underline file writer in Iceberg.

That is the reason why @huyuanfeng2018 was suggesting if RowData is the only key type we need to support. If in the future, Flink sink supports other record format for the underline file writer, maybe we can revisit this decision. for now, RowData is the only type Flink file writer supports.

Thanks @stevenzwu for the answer!

[yegangy0718]

One more question, how do we plan to register the DataStatisticsOrRecordSerializer? Do we plan to define TypeInfoFactory or use env config to register it?

[stevenzwu]

One naive question, how do we plan to register the DataStatisticsOrRecordSerializer? Do we plan to define TypeInfoFactory or use env config to register it?

@yegangy0718 good question. we probably also need to implement a TypeInfo class when adding the DataStatisticsOperator. that way, Flink would know which type serializer to use. TypeInformation impl can be followed up as a separate PR.

Here is the DataStream API.

    /**
     * Method for passing user defined operators along with the type information that will transform
     * the DataStream.
     *
     * @param operatorName name of the operator, for logging purposes
     * @param outTypeInfo the output type of the operator
     * @param operator the object containing the transformation logic
     * @param <R> type of the return stream
     * @return the data stream constructed
     * @see #transform(String, TypeInformation, OneInputStreamOperatorFactory)
     */
    @PublicEvolving
    public <R> SingleOutputStreamOperator<R> transform(
            String operatorName,
            TypeInformation<R> outTypeInfo,
            OneInputStreamOperator<T, R> operator) {

        return doTransform(operatorName, outTypeInfo, SimpleOperatorFactory.of(operator));
    }