如何在Flink中使用LocalDate字段高效地序列化POJO？

Question

org.apache.flink.api.java.typeutils.TypeExtractor - Class class java.time.LocalDate cannot be used as a POJO type because not all fields are valid POJO fields, and must be processed as GenericType. Please read the Flink documentation on "Data Types & Serialization" for details of the effect on performance.

Answer 1

由于向后兼容性，即使Flink中引入了新的序列化程序，也不能自动使用它。但是，您可以这样告诉Flink将其用于您的POJO（如果您使用Kryo在没有先前保存点的情况下开始）：

@TypeInfo(MyClassTypeInfoFactory.class)
public class MyClass {
  public int id;
  public LocalDate date;
  // ...
}

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

    return Types.POJO(MyClass.class, new HashMap<String, TypeInformation<?>>() { {
        put("id", Types.INT);
        put("date", Types.LOCAL_DATE);
        // ...
    } } );
  }
}

您必须为POJO的所有字段提供类型，如图所示，但是types类中有很多帮助器可以使用。此外，通过像这样使用typeInfoFactory，您不必担心Flink使用该类型的所有位置-它总是派生给定的类型信息。

如果需要转换旧的保存点以使用新的序列化程序，您可能还需要查看Flink的状态处理器API。