4. Flink数据类型与序列化：支持的数据类型、TypeInformation与自定义Serializer

在实时特征计算中，数据在算子之间传输、状态后端持久化、以及跨网络Shuffle时，都必须经过序列化与反序列化。Flink内置了一套高效的类型系统，能够自动推断数据类型并生成最优序列化器，避免Java/Scala原生序列化的性能开销。理解Flink的数据类型体系与序列化机制，是编写高性能实时特征作业的基础。

4.1 支持的数据类型体系

Flink将数据类型分为四大类别，每一类都有对应的序列化策略和TypeInformation表示。

4.2 TypeInformation：类型系统的核心

TypeInformation 是Flink类型系统的根接口，它描述了数据类型的所有元信息，并负责生成对应的 TypeSerializer。Flink通过 TypeExtractor 或 TypeHint 自动推断 TypeInformation，开发者也可以在 returns() 方法中显式指定。

4.2.1 常见TypeInformation实现

• BasicTypeInfo：对应基础类型，如 INT_TYPE_INFO、STRING_TYPE_INFO。
• TupleTypeInfo：描述Tuple类型，包含每个字段的TypeInformation。
• PojoTypeInfo：描述POJO类型，通过反射获取所有公有字段及其类型。
• CaseClassTypeInfo：Scala Case Class专用，保留字段顺序和名称。
• GenericTypeInfo：当Flink无法识别时回退到Kryo序列化，性能较差，应尽量避免。

// Java: 显式指定TypeInformation
DataStream<UserBehavior> stream = env
    .addSource(kafkaSource)
    .returns(TypeInformation.of(UserBehavior.class));

// Scala: 使用TypeHint
val stream: DataStream[(Long, String, Double)] = env
    .fromElements((1L, "click", 0.5))
    .returns(TypeInformation.of(new TypeHint[(Long, String, Double)] {}))

4.3 POJO类型的要求与最佳实践

Flink对POJO类型有严格的定义要求，满足以下条件的类才会被识别为POJO，从而使用高效的POJO序列化器：

1. 类是 public 的。
2. 有一个 无参构造方法（public）。
3. 所有字段都是 public 的，或者提供了符合Java Bean规范的 getter/setter 方法。
4. 字段类型必须是Flink支持的序列化类型。

// 符合Flink POJO规范的示例
public class FeatureEvent {
    public Long userId;          // public字段
    public String featureName;
    public Double featureValue;
    public Long timestamp;

    // 必须有无参构造
    public FeatureEvent() {}

    // 全参构造可选
    public FeatureEvent(Long userId, String featureName, Double featureValue, Long timestamp) {
        this.userId = userId;
        this.featureName = featureName;
        this.featureValue = featureValue;
        this.timestamp = timestamp;
    }
}

4.4 自定义TypeSerializer

当内置序列化器无法满足性能要求或特殊格式需求时（如压缩、加密、自定义二进制协议），可以实现 TypeSerializer 接口来定制序列化逻辑。自定义Serializer通常与 TypeInformation 配合使用，通过 createSerializer() 方法返回。

4.4.1 核心接口方法

// 自定义Serializer示例：压缩存储特征值
public class CompressedDoubleSerializer extends TypeSerializer<Double> {

    @Override
    public void serialize(Double value, DataOutputView target) throws IOException {
        // 将Double转为int表示（乘以1000取整），减少体积
        int compressed = (int) (value * 1000);
        target.writeInt(compressed);
    }

    @Override
    public Double deserialize(DataInputView source) throws IOException {
        int compressed = source.readInt();
        return compressed / 1000.0;
    }

    @Override
    public TypeSerializer<Double> duplicate() {
        return new CompressedDoubleSerializer();
    }

    // 其他方法实现略...
}

4.5 类型系统与序列化架构

下图展示了Flink类型系统从数据源到算子执行的全链路序列化流程：

4.6 类型选择决策流程

在实际开发中，选择合适的数据类型和序列化方式可以遵循以下决策树：

1. 是否为基础类型或简单包装？ → 直接使用 BasicTypeInfo，无需额外配置。
2. 是否为固定字段组合且字段数≤25？ → 使用 Tuple，适合临时中间结果。
3. 是否需要语义清晰的字段名和嵌套结构？ → 定义 POJO（Java）或 Case Class（Scala），这是特征实体的首选。
4. 是否有特殊序列化需求（压缩、加密、自定义协议）？ → 实现 TypeSerializer 接口，并通过 TypeInformation 注册。
5. 是否使用了泛型或Lambda导致类型擦除？ → 必须使用 .returns() 或 TypeHint 显式指定。