Flink 的接口的 POJO 序列化器

Question

假设我们有一个接口 Shape 及其实现的人为示例。我如何序列化包含形状的 myPojo （使用 POJO 序列化程序）？ Circle 和 Rectangle 都是 POJO。

public interface Shape {
    public double area(); 
}

public class Circle implements Shape{
    // constructors
    // radius
    // implement area();
    // getters, setters
}

public class Rectangle implements Shape {
    // constructors
    // height, width
    // implement area();
    // getters, setters
}

public class MyPojo {
    int anotherField;
    Shape shape;
    // constructors
   // getters, setters
}

我的执行配置如下所示：

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.getConfig().disableGenericTypes();
    env.getConfig().registerPojoType(Shape.class);
    env.getConfig().registerPojoType(Circle.class);
    env.getConfig().registerPojoType(Rectangle.class);

Answer 1

Shape 没有有效的方法来考虑 POJO

All fields are either public or must be accessible through getter and setter functions. For a field called foo the getter and setter methods must be named getFoo() and setFoo().

public interface Shape {
    public double getArea(); 
    public void setArea(double area); 
}

Answer 2

问题是 Shape 接口没有字段，即使创建一个虚拟字段被认为是静态的，因此具有相同的结果。我得出的“hack”解决方案是为字段提供一个空的 HashMap：

public class ShapeTypeInfoFactory extends TypeInfoFactory<Shape> {
  @Override
  public TypeInformation<Shape> createTypeInfo(
      Type t, Map<String, TypeInformation<?>> genericParameters) {
    return Types.POJO(Shape.class, new HashMap<>());
  }
}

Flink 的接口的 POJO 序列化器

Flink's POJO serializer for an interface

java

apache-flink