前言

在传统的批处理中，数据划分为一个个batch，然后每一个Task去处理一个batch。一个批次的数据通过计算处理输出就是最终的结果。对于state的需求不高

对于流计算而言，对State有非常高的要求，特别在实时流中，输入是一个无限制的流，服务开启后基本都不会停机。在生产中，大部分需求都是有状态的计算就需要将状态数据很好的管理起来。

在传统的流计算系统中，对状态管理支持并不是很完善。

FLink提供了丰富的状态访问和容错机制。

这篇文章主要介绍说明各种State，并介绍如何在Flink应用程序中使用状态。关于checkpoint机制和容错机制详细内容会在后面文章详细说明。

State

What is State

大多数流应用程序都是有状态的。许多算子会不断地读取和更新状态
例如worldcount程序，也要读取当前的状态去计算再更新状态。

一般来说，由一个任务维护，并且用来计算某个结果的所有数据，都属于这个任务的状态。

你可以认为状态就是一个本地变量，可以被任务的业务逻辑访问。

image.png

任务会接收一些输入数据。在处理数据时，任务可以读取和更新状态，并根据输入数据和状态计算结果。

应用程序里，读取和写入状态的逻辑一般都很简单直接，而有效可靠的状态管理会复杂一些。这包括如何处理很大的状态——可能会超过内存，并且保证在发生故障时不会丢失任何状态。

Flink会帮我们处理这相关的所有问题，包括状态一致性、故障处理以及高效存储和访问，以便开发人员可以专注于应用程序的逻辑。

Flink有两种类型的状态：算子状态（operator state）和键控状态（keyed state）

算子状态（operator state or non-keyed state)

顾名思义，状态是和算子进行绑定的，一个算子的状态不能被其他算子所访问到。状态由同一并行任务所处理的所有数据都可以访问到相同的状态，状态对于同一并行任务而言是共享的。算子状态不能由相同或不同算子的另一个任务访问。

operator state

Kafka Connector是一个很好的operator state 使用示例。
Kafka consumer 的每个并行实例都维护一个topic partition和offset的映射作为其operator state 。
当更改并行时，operator state 接口支持在并行的实例重新分配state。有多种执行此重新分配的方案。

算子状态的类型

算子状态三种基本数据结构：

BroadcastState 用于广播的算子状态。
BroadcastState是 Operator State的一种特殊类型。引入它是为了支持需要将一个流的记录广播到所有下游任务的用例，这些记录用于在所有子任务之间保持相同的状态。

ListState 将状态表示为一组数据的列表。

UnionListState 存储列表类型的状态，与 ListState 的区别在于：如果并行度发生变化，ListState 会将该算子的所有并发的状态实例进行汇总，然后均分给新的 Task；而 UnionListState 只是将所有并发的状态实例汇总起来，具体的划分行为则由用户进行定义。

算子状态的使用

Flink_operator_state-iteblog

键控状态（keyed state）

顾名思义，键控状态是根据输入数据流中定义的键（key）来维护和访问的。

Flink为每个键值维护一个状态实例，并将具有相同键的所有数据，都分区到同一个算子任务中，这个任务会维护和处理这个key对应的状态。

你可以将Keyed State视为是已经被分片或分区的Operator State，每个key都有且仅有一个状态分区(state-partition)。

每个keyed-state逻辑上绑定到一个唯一的<parallel-operator-instance, key>组合上，由于每个key“属于”keyed operator的一个并行实例，所以我们可以简单的认为是<operator,key>。

Keyed State进一步被组织到所谓的Key Groups中。Key Groups是Flink能够重新分配keyed State的原子单元。Key Groups的数量等于定义的最大并行度。在一个keyed operator的并行实例执行期间，它与一个或多个Key Groups配合工作。

当任务处理一条数据时，它会自动将状态的访问范围限定为当前并行以及是当前数据的key。
因此，具有相同并行和相同的key的所有数据都会访问相同的状态。

下图显示了任务如何与键控状态进行交互。

keyed state

键控状态的类型

键控状态接口提供对不同类型的状态的访问，所有状态都根据输入数据流中定义的键（key）来维护和访问的。这意味着这种状态类型只能在KeyedStream上使用

不同类型的键控状态：

• ValueState[T]保存单个的值，值的类型为T。
  get操作: ValueState.value()
  set操作: ValueState.update(value: T)

• ListState[T]保存一个列表，列表里的元素的数据类型为T。基本操作如下：
  ListState.add(value: T)
  ListState.addAll(values: java.util.List[T])
  ListState.get()返回Iterable[T]
  ListState.update(values: java.util.List[T])

• MapState[K, V]保存Key-Value对
  MapState.get(key: K)
  MapState.put(key: K, value: V)
  MapState.contains(key: K)
  MapState.remove(key: K)

• ReducingState[T] 用于存储经过 ReduceFunction 计算后的结果，接口和ListState相同，使用 add(T) 增加元素，但是使用add(T)方法本质是使用指定ReduceFunction的聚合行为。

• AggregatingState[I, O] 它保存了一个聚合了所有添加到这个状态的值的结果。与ReducingState有些不同，聚合类型可能不同于添加到状态的元素的类型。接口和ListState相同，但是使用add(IN)添加的元素本质是通过使用指定的AggregateFunction进行聚合。

所有类型的状态都具有clear()清除当前活动键（即输入元素的键）状态的方法。

注意：

1. 这些状态对象只能用来与状态进行交互。状态不一定存储在内存中，但是可能存储在磁盘或者DB中。这依赖于状态后端的设置。
2. 从状态获取的值依赖于输入元素的key。因此如果包含不同的key，那么在你的用户函数中的一个调用获得的值和另一个调用获得值可能不同。

键控状态的使用

Flink_keyed_state-iteblog

为了获得状态句柄，必须创建一个StateDescriptor。它维护了状态的名称，状态维护的值的类型，和可用户定义function（ValueState、ListState），例如ReduceFunction。根据你想要查询的状态的类型，你可以创建ValueStateDescriptor，ListStateDescriptor，ReducingStateDescriptor，FoldingStateDescriptor或MapStateDescriptor。

通过RuntimeContext注册StateDescriptor。StateDescriptor以状态state的名字和存储的数据类型为参数。数据类型必须指定，因为Flink需要选择合适的序列化器。

lazy val listState = getRuntimeContext.getListState(
      new ListStateDescriptor[SensorReading]("list-state", Types.of[SensorReading])
)

当一个函数注册了StateDescriptor描述符，Flink会检查状态后端是否已经存在这个状态。这种情况通常出现在应用挂掉要从检查点或者保存点恢复的时候。在这两种情况下，Flink会将注册的状态连接到已经存在的状态。如果不存在状态，则初始化一个空的状态。

状态生存时间（TTL）

任何类型的 keyed state 都支持配置有效期 (TTL) 。如果配置了TTL且状态值已过期，则将清除存储的值

为了使用状态TTL，必须首先构建一个StateTtlConfig配置对象。然后可以通过传递配置在任何状态描述符中启用TTL功能：

import org.apache.flink.api.common.state.StateTtlConfig
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.api.common.time.Time

val ttlConfig = StateTtlConfig
    .newBuilder(Time.seconds(1))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .build
    
val stateDescriptor = new ValueStateDescriptor[String]("text state", classOf[String])
stateDescriptor.enableTimeToLive(ttlConfig)

newBuilder 方法的参数是状态的有效期。

setUpdateType 配置何时刷新状态TTL（默认为OnCreateAndWrite）

• StateTtlConfig.UpdateType.OnCreateAndWrite -仅在创建和写访问权限时
• StateTtlConfig.UpdateType.OnReadAndWrite -也具有读取权限

setStateVisibility 配置是否清除尚未过期的默认值（默认情况下NeverReturnExpired）

• StateTtlConfig.StateVisibility.NeverReturnExpired -永不返回过期值
• StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp -如果仍然可用，则返回

note:

• 启用此功能会增加state backend 状态存储的消耗
• 当前仅支持有关processing time TTL 。
• TTL配置不是checkpoint或savepoint的一部分

operator state 和 keyed state区别

接下来我们会在四个维度来区分两种不同的 state：operator state 以及 keyed state。

1. 是否存在当前处理的 key（current key）：operator state 是没有当前 key 的概念，而 keyed state 的数值总是与一个 current key 对应。
2. 存储对象是否 on heap: 目前 operator state backend 仅有一种 on-heap 的实现；而 keyed state backend 有 on-heap 和 off-heap（RocksDB）的多种实现。
3. 是否需要手动声明快照（snapshot）和恢复 (restore) 方法：operator state 需要手动实现 snapshot 和 restore 方法；而 keyed state 则由 backend 自行实现，对用户透明。
4. 数据大小：一般而言，我们认为 operator state 的数据规模是比较小的；认为 keyed state 规模是相对比较大的。需要注意的是，这是一个经验判断，不是一个绝对的判断区分标准。

State Persistence 状态持久化

Flink通过结合stream replay 和 checkpointing来实现容错。checkpoint标记每个输入流中的特定点以及每个运算符的对应state。通过并从检查点恢复state保持一致性（一次处理语义）。

如果发生程序故障（由于机器，网络或软件故障），Flink将停止分布式流数据流。然后，系统重新启动操作员，并将其重置为最新的成功检查点。输入流将重置为状态快照的点。确保作为重新启动的并行数据流的一部分处理的任何记录都不会影响以前的检查点状态。

note:

• 默认情况下，检查点是禁用的。
• 为了使该机制实现其全部保证，数据流源（例如消息队列）必须能够将流回滚到定义的offset位置。kafka connector是利用这一功能实现的。比如socket source没有回滚消费记录功能，无法实现。
• 由于Flink的检查点是通过分布式快照实现的，snapshot 和checkpoint 交替使用。通常，我们也使用snapshot 一词来表示checkpoint 或savepoint

StateBackend 状态后端

每传入一条数据，有状态的算子任务都会读取和更新状态。当配置了检查点时，此状态会在检查点上保持不变，以防止数据丢失并持续恢复。状态如何在内部表示以及在检查点上如何保持以及在何处持久取决于所选的State Backend。

由于有效的状态访问对于处理数据的低延迟至关重要，因此每个并行任务都会在本地维护其状态，以确保快速的状态访问

状态后端主要负责两件事：本地的状态管理，以及将检查点（checkpoint）状态写入远程存储

状态后端的选择

MemoryStateBackend

内存级的状态后端，会将键控状态作为内存中的对象进行管理，将它们存储在 TaskManager 的 JVM 堆上，而将 checkpoint 存储在 JobManager 的内存中

可以将MemoryStateBackend配置为使用异步快照。

new MemoryStateBackend(MAX_MEM_STATE_SIZE, false);

其中，new MemoryStateBackend(DEFAULT_MAX_STATE_SIZE,false) 中的 false 代表关闭异步快照机制

MemoryStateBackend一般用来进行本地调试用，使用 MemoryStateBackend 时需要注意的一些点包括：

• 每个独立的状态（state）默认限制大小为 5MB，可以通过构造函数增加容量
• 状态的大小不能超过 akka 的 Framesize 大小
• 聚合后的状态必须能够放进 JobManager 的内存中

适用情况：

• 适用于我们本地调试使用，来记录一些状态很小的 Job 状态信息。
• 几乎没有状态的作业，例如仅包含一次记录功能（Map，FlatMap，Filter等）的作业。kafka consumer 只需要很少的状态。

FsStateBackend

FsStateBackend 会把状态数据保存在 TaskManager 的内存中。CheckPoint 时，将状态快照写入到配置的文件系统目录中，少量的元数据信息存储到 JobManager 的内存中。

使用 FsStateBackend 需要我们指定一个文件路径，一般来说是 HDFS 的路径，比如：“hdfs://node09:8020/flink/checkpoints” or “file:///data/flink/checkpoints”

构造函数为：

new FsStateBackend(path, false);

第二个参数false 也是代表关闭异步快照机制

适用情况：

• 状态较大，窗口较长，键/值状态较大的作业。

• 所有高可用性设置。

RocksDBStateBackend

与 FsStateBackend 不同的是，RocksDBStateBackend 将正在运行中的状态数据保存在 RocksDB 数据库中，RocksDB 数据库默认将数据存储在 TaskManager 运行节点的数据目录下。

与 FsStateBackend 类似的他们都需要一个外部文件存储路径，比如：“hdfs://node09:8020/flink/checkpoints” or “file:///data/flink/checkpoints”

RocksDBStateBackend始终执行异步快照。

RocksDBStateBackend的局限性：

• 由于RocksDB的JNI bridge API基于byte []，因此每个键和每个值的最大支持大小为2 ^ 31个字节。 重要说明：在RocksDB中使用合并操作的状态（例如ListState）可以静默累积值大于 2 ^ 31字节，在下一次数据检索时将会失败。目前，这是RocksDB JNI的限制。

适用情况：

• 状态很大，窗口较长，键/值状态很大的作业。

• 所有高可用性设置。

与 FsStateBackend 不同的是，RocksDBStateBackend 将正在运行中的状态数据保存在 RocksDB 数据库中，RocksDB 数据库默认将数据存储在 TaskManager 运行节点的数据目录下。

这意味着，RocksDBStateBackend 可以存储远超过 FsStateBackend 的状态，可以避免向 FsStateBackend 那样一旦出现状态暴增会导致 OOM，但是因为将状态数据保存在 RocksDB 数据库中，吞吐量会有所下降。

状态检查点（checkpoint）的写入也非常重要，这是因为Flink是一个分布式系统，而状态只能在本地维护。 TaskManager进程（所有任务在其上运行）可能在任何时间点挂掉。因此，它的本地存储只能被认为是不稳定的。状态后端负责将任务的状态检查点写入远程的持久存储。写入检查点的远程存储可以是分布式文件系统，也可以是数据库。不同的状态后端在状态检查点的写入机制方面有所不同。RocksDBStateBackend 是唯一支持增量快照的状态后端，这对于非常大的状态来说，可以显著减少状态检查点写入的开销。

状态后端的配置

没有指定的时候，默认是使用jobmanager（MemoryStateBackend）。如果希望为群集上的所有作业建立不同的默认值，则可以通过在flink-conf.yaml中定义新的默认状态后端来实现。可以按作业覆盖默认状态后端，如下所示。

设置默认状态后端

可以flink-conf.yaml 通过修改key为 state.backend 的value值。具体的value值以及对应的状态后端

jobmanager（MemoryStateBackend），filesystem（FsStateBackend），rocksdb（RocksDBStateBackend）

或者自定义class实现StateBackendFactory 接口 写上class的完全限定类名，例如org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactoryRocksDBStateBackend

配置文件中的示例部分可能如下所示：

# The backend that will be used to store operator state checkpoints

state.backend: filesystem


# Directory for storing checkpoints

state.checkpoints.dir: hdfs://namenode:40010/flink/checkpoints

设置每个job的状态后端

可以通过StreamExecutionEnvironment 指定job的状态后端，如下例所示：

val env = StreamExecutionEnvironment.getExecutionEnvironment()
env.setStateBackend(new FsStateBackend("hdfs://node09:8020/flink/checkpoints"))

如果要在IDE中使用RocksDBStateBackend必须将以下依赖项添加到Flink项目中。

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
    <version>1.11.0</version>
    <scope>provided</scope>
</dependency>

注意：由于RocksDB是默认Flink发行版的一部分，因此，如果您在工作中未使用任何RocksDB代码，则不需要此依赖项。

参考FLink 官网State & Fault Tolerance
参考Flink State 最佳实践
参考过往记忆的博客Apache Flink状态管理和容错机制介绍