Flink

状态模式的演变

Sat Aug 22, 2020

State Schema Evolution

Apache Flink 流媒体应用通常被设计为无限期或长时间运行。与所有长期运行的服务一样，应用程序需要更新以适应不断变化的需求。这对于应用程序所针对的数据模式(data schema)也是一样的，它们会随着应用程序的发展而发展。

本页提供了关于如何演进状态类型的数据模式(data schema)的概述。当前的限制在不同的类型和状态结构（ValueState、ListState 等）中有所不同。

请注意，本页面上的信息仅在您使用由 Flink 自己的类型序列化框架生成的状态序列化器时相关。也就是说，在声明你的状态时，所提供的状态描述符并没有被配置为使用特定的 TypeSerializer 或 TypeInformation，在这种情况下，Flink 会推导出状态类型的信息。

ListStateDescriptor<MyPojoType> descriptor =
    new ListStateDescriptor<>(
        "state-name",
        MyPojoType.class);

checkpointedState = getRuntimeContext().getListState(descriptor);

在底层，状态的模式(schema)是否可以被演化取决于用于读取/写入持久化状态字节的序列化器。简单地说，只有当它的序列化器正确地支持时，一个注册状态的模式才能被演化。这是由 Flink 的类型序列化框架生成的序列化器透明地处理的（当前的支持范围列在下面）。

如果你打算为你的状态类型实现一个自定义的 TypeSerializer，并想了解如何实现序列化器以支持状态模式演化，请参考自定义状态序列化。那里的文档还涵盖了关于状态序列化器和 Flink 的状态后端之间的相互作用的必要内部细节，以支持状态模式(state schema)演化。

状态模式的演化 #

要演化给定状态类型的模式，您需要采取以下步骤。

保存你的 Flink 流作业(job)的保存点。
更新您的应用程序中的状态类型（例如，修改您的 Avro 类型模式）。
从保存点恢复作业(job)。当第一次访问状态时，Flink 将评估是否已经改变了状态的模式(schema)，并在必要时迁移状态模式。

迁移状态以适应已更改的模式的过程是自动发生的，并且对每个状态都是独立的。这个过程由 Flink 内部执行，首先检查状态的新序列器是否与之前的序列器有不同的序列化模式，如果有，则用之前的序列器将状态读到对象，再用新的序列器写回字节。

关于迁移过程的进一步细节不在本文档的范围内，请参考这里。

支持的模式演化数据类型 #

目前，模式演化只支持 POJO 和 Avro 类型。因此，如果你关心状态的模式演化，目前建议始终使用 POJO 或 Avro 作为状态数据类型。

有计划扩展对更多复合类型的支持；更多细节请参考 FLINK-10896。

POJO 类型 #

Flink 支持 POJO 类型的演化模式，基于以下一组规则。

字段可以被删除。一旦被删除，在未来的检查点和保存点中，被删除字段的之前值将被丢弃。
可以添加新字段。新字段将被初始化为其类型的默认值，正如 Java 所定义的那样。
已声明的字段类型不能改变。
POJO 类型的类名不能改变，包括类的命名空间。

请注意，POJO 类型状态的模式只能在 Flink 版本大于 1.8.0 的情况下，从以前的保存点恢复时才能进化。当使用比 1.8.0 更老的 Flink 版本进行还原时，模式不能被改变。

Avro 类型 #

Flink 完全支持 Avro 类型状态的演变模式，只要模式变化被 Avro 的模式解析规则认为是兼容的。

一个限制是作为状态类型使用的 Avro 生成的类在恢复作业时不能被重新定位或具有不同的命名空间。

注意: 不支持键的模式演变。

举个例子。RocksDB 状态后端依赖于二进制对象的标识，而不是 hashCode 方法实现。对 keys 对象结构的任何改变都可能导致非确定性行为。

注意: Kryo 不能用于模式演化。

当使用 Kryo 时，框架没有可能验证是否有任何不兼容的变化。

原文链接: https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/stream/state/schema_evolution.html