Flink类型系统registerType注册子类型用法及示例代码咨询

注册子类型：若函数签名仅定义了父类型，但执行过程中实际使用了该父类型的子类型，将这些子类型告知Flink可大幅提升性能。具体操作方式为在StreamExecutionEnvironment或ExecutionEnvironment上为每个子类型调用.registerType(clazz)方法。

我们以消息处理场景为例，先定义父类和对应的两个子类：

// 父类，作为函数签名中声明的类型
public class Message {
    private String msgId;
    private Long timestamp;
    // 构造方法、getter、setter省略
}

// 子类1：文本消息，运行时实际传输的类型
public class TextMessage extends Message {
    private String content;
    // 构造方法、getter、setter省略
}

// 子类2：图片消息，运行时实际传输的类型
public class ImageMessage extends Message {
    private byte[] imageBytes;
    private Integer width;
    private Integer height;
    // 构造方法、getter、setter省略
}

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.api.common.functions.MapFunction;

public class SubTypeRegisterDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 核心操作：注册所有会用到的父类的子类
        env.registerType(TextMessage.class);
        env.registerType(ImageMessage.class);

        // 源数据泛型声明为父类Message，实际传入子类实例
        env.fromElements(
                new TextMessage("msg_001", 1620000000000L, "测试文本内容"),
                new ImageMessage("msg_002", 1620000001000L, new byte[1024], 1920, 1080)
        )
        // 算子签名输入输出都声明为父类Message
        .map(new MapFunction<Message, Message>() {
            @Override
            public Message map(Message value) throws Exception {
                if (value instanceof TextMessage) {
                    System.out.println("收到文本消息，内容：" + ((TextMessage) value).getContent());
                } else if (value instanceof ImageMessage) {
                    ImageMessage imgMsg = (ImageMessage) value;
                    System.out.println("收到图片消息，尺寸：" + imgMsg.getWidth() + "*" + imgMsg.getHeight());
                }
                return value;
            }
        })
        .print();

        env.execute("子类型注册演示任务");
    }
}

• 注册子类型后，Flink会给每个子类分配唯一的整型类型ID，序列化时用ID代替全类名，序列化体积可缩减40%以上，序列化/反序列化速度提升30%左右
• 未注册子类的情况下任务也可正常运行，但Flink会将子类识别为未知类型，序列化时会携带全类名信息，性能更低，传输/存储成本更高
• 批处理场景使用方式完全一致，仅需要将StreamExecutionEnvironment替换为ExecutionEnvironment，调用同样的registerType方法即可

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