Apache Flink含Tuple列表的POJO类型信息错误求助

问题说明 #

无法创建包含Tuple列表数据类型的POJO类，疑似类型信息配置错误。执行DataStream的map操作时抛出类型转换异常；禁用泛型序列化env.getConfig().disableGenericTypes()后可正常运行，确认问题出在POJO序列化的类型信息配置环节。

相关代码 #

1. NodeTest POJO类 #

@TypeInfo(NodeTestSerializer.class)
public class NodeTest implements Serializable {
    public List<Tuple3<Integer,Integer, String>> msgDestinationList = new ArrayList<>();

    public NodeTest() {
        msgDestinationList.add(Tuple3.of(1,2,"abc"));
    };
}

2. 自定义TypeInfoFactory实现 #

public class NodeTestSerializer extends TypeInfoFactory<NodeTest> {
    @Override
    public TypeInformation<NodeTest> createTypeInfo(Type t, Map<String, TypeInformation<?>> genericParameters) {

        TypeInformation<Tuple3<Integer,Integer, String>> tupleTypeInfo =
                TypeInformation.of(new TypeHint<Tuple3<Integer, Integer, String>>() {});

        ImmutableMap nMap = ImmutableMap.builder()
                .put("msgDestinationList", Types.LIST(Types.TUPLE(tupleTypeInfo)))
                .build();

        final TypeInformation<NodeTest> nodeInfo = Types.POJO(NodeTest.class,nMap);
        return nodeInfo;
    }
}

3. 主执行类 #

public class NodeTestMain {

    public static void main(final String[] args) throws IOException {
        try {

            final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.getConfig().disableGenericTypes();   // 禁用泛型序列化（Kryo）

            DataStream<NodeTest> nodeStream = env.fromElements(new NodeTest());
            nodeStream.map(i -> i.msgDestinationList.get(0).f0).print();

            env.execute("Flink Node Test");
        }
        catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        }
    }

}

错误日志 #

java.io.UncheckedIOException: java.io.IOException: Serializing the source elements failed: class java.lang.Integer cannot be cast to class org.apache.flink.api.java.tuple.Tuple (java.lang.Integer is in module java.base of loader 'bootstrap'; org.apache.flink.api.java.tuple.Tuple is in unnamed module of loader 'app')
    at org.apache.flink.streaming.api.functions.source.FromElementsFunction.setOutputType(FromElementsFunction.java:158)
    ...
Caused by: java.lang.ClassCastException: class java.lang.Integer cannot be cast to class org.apache.flink.api.java.tuple.Tuple (java.lang.Integer is in module java.base of loader 'bootstrap'; org.apache.flink.api.java.tuple.Tuple is in unnamed module of loader 'app')
    at org.apache.flink.api.java.typeutils.runtime.TupleSerializer.serialize(TupleSerializer.java:37)
    ...

解决方案 #

问题出在自定义TypeInfoFactory中对Tuple类型的错误构造：Types.TUPLE(tupleTypeInfo)用法有误，Types.TUPLE()需要传入Tuple每个字段的TypeInformation，而非整个Tuple的类型信息。

修改后的NodeTestSerializer代码如下：

public class NodeTestSerializer extends TypeInfoFactory<NodeTest> {
    @Override
    public TypeInformation<NodeTest> createTypeInfo(Type t, Map<String, TypeInformation<?>> genericParameters) {
        // 直接构造Tuple3的类型信息
        TypeInformation<Tuple3<Integer, Integer, String>> tupleTypeInfo =
                TypeInformation.of(new TypeHint<Tuple3<Integer, Integer, String>>() {});
        
        // 列表类型直接传入Tuple的类型信息，无需用Types.TUPLE()包装
        ImmutableMap<String, TypeInformation<?>> nMap = ImmutableMap.builder()
                .put("msgDestinationList", Types.LIST(tupleTypeInfo))
                .build();

        return Types.POJO(NodeTest.class, nMap);
    }
}

原理说明 #

Types.TUPLE()用于动态构造不确定字段数量的Tuple类型，参数是每个字段的类型信息；对于确定类型的Tuple（如Tuple3），直接通过TypeInformation.of(TypeHint)就能获取正确的类型信息，传入Types.LIST()后可让Flink正确识别列表中的Tuple类型，避免序列化时的类型转换错误。

内容的提问来源于stack exchange，提问作者K.M