如何为Flink累加器对象提供类型信息以避免使用Kryo序列化

你遇到的问题是因为Java泛型擦除机制，Flink的TypeExtractor无法自动推断出AggregationAccumulator中customMap字段的泛型参数类型（String和Long），因此会将该字段标记为GenericType，转而使用Kryo进行序列化，这会影响性能和 schema 演进能力。以下是两种可靠的解决方式：

方式一：重写getAccumulatorType方法显式指定类型信息 #

在你的CustomAggregateFunction中重写getAccumulatorType方法，通过Flink的Types工具类明确指定累加器的POJO类型，以及customMap字段的具体泛型类型：

public class CustomAggregateFunction implements AggregateFunction<CustomInput, AggregationAccumulator, CustomOutput> {

    // 实现AggregateFunction的其他方法（createAccumulator、add、getResult、merge）

    @Override
    public TypeInformation<AggregationAccumulator> getAccumulatorType() {
        return Types.POJO(AggregationAccumulator.class,
                Map.of("customMap", Types.MAP(Types.STRING, Types.LONG)));
    }

    @Data
    public static class AggregationAccumulator {
        private Map<String, Long> customMap = new HashMap<>();
    }
}

这种方式直接给Flink提供了明确的类型元数据，避免了自动推断的不确定性。

方式二：使用@TypeHint注解辅助类型推断 #

在AggregationAccumulator的customMap字段上添加@TypeHint注解，明确标注泛型参数的具体类型，帮助TypeExtractor正确识别：

public class CustomAggregateFunction implements AggregateFunction<CustomInput, AggregationAccumulator, CustomOutput> {

    // 实现AggregateFunction的其他方法

    @Data
    public static class AggregationAccumulator {
        @TypeHint(typeInfo = MapTypeInfo.class, parameters = {String.class, Long.class})
        private Map<String, Long> customMap = new HashMap<>();
    }
}

或者也可以用字符串形式指定完整类型：

@TypeHint("java.util.Map<java.lang.String, java.lang.Long>")
private Map<String, Long> customMap = new HashMap<>();

两种方式都能让Flink识别出customMap的具体类型，从而使用Flink原生的序列化器替代Kryo，解决提示问题。

内容的提问来源于stack exchange，提问作者Baiqing