Flink 的接口 POJO 序列化器

Question

假设我们有一个人为的接口Shape示例及其实现。如何序列化包含形状的myPojo(使用POJO序列化器)?圆形和矩形都是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的有效方法

所有字段要么是公共的，要么必须通过getter和setter函数访问。对于名为foo的字段，getter和setter方法必须命名为getFoo()和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<>());
}
}