rakulang, dartlang, nimlang, golang, rustlang, lang lang no see

状态模式的演变

焉知非鱼

State Schema Evolution

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

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

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

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

checkpointedState = getRuntimeContext().getListState(descriptor);

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

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

状态模式的演化 #

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

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

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

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

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

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

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

POJO 类型 #

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

  1. 字段可以被删除。一旦被删除,在未来的检查点和保存点中,被删除字段的之前值将被丢弃。
  2. 可以添加新字段。新字段将被初始化为其类型的默认值,正如 Java 所定义的那样。
  3. 已声明的字段类型不能改变。
  4. 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