如何为Flink MapState指定POJO类型ArrayList的Value Class Type？

当你需要声明值类型为List<Pojo>的MapState时，直接传入List.class或Pojo.class都无法工作，根源是Java泛型擦除导致Flink无法识别List内部的元素类型，以及类型不匹配问题。以下是两种可行的解决方案：

方案1：使用TypeHint（推荐，支持复杂泛型场景） #

通过TypeHint可以保留泛型的类型信息，让Flink明确知道List的元素是你的POJO类：

private transient MapState<String, List<Pojo>> mapState;

mapState = getRuntimeContext().getMapState(
        new MapStateDescriptor<>(
                "RIGHT_BUFFER",
                String.class,
                TypeInformation.of(new TypeHint<List<Pojo>>() {})
        )
);

方案2：使用ListTypeInfo（专门针对List类型） #

Flink提供了ListTypeInfo类，直接指定内部元素的类型即可：

private transient MapState<String, List<Pojo>> mapState;

mapState = getRuntimeContext().getMapState(
        new MapStateDescriptor<>(
                "RIGHT_BUFFER",
                String.class,
                new ListTypeInfo<>(Pojo.class)
        )
);

额外注意事项 #

你的Pojo类必须符合Flink的POJO序列化要求：

• 拥有公共的无参构造方法
• 所有字段要么是公共访问权限，要么提供对应的getter和setter方法
• 字段类型必须是Flink支持的序列化类型（比如基本类型、String、其他符合要求的POJO等）

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