上一篇文章我们谈论了保存点的相关内容，其中就谈到了保存点状态的存储。这篇文章我们来探讨用户程序状态的存储，也是在之前的文章中多次提及的state backend(中文暂译为状态终端)。

基于数据流API而编写的程序经常以各种各样的形式保存着状态：

• 窗口收集/聚合元素（这里的元素可以看作是窗口的状态）直到它们被触发
• 转换函数可能会使用key/value状态接口来存储数据
• 转换函数可能实现Checkpointed接口来让它们的本地变量受益于fault tolerant机制

当检查点机制工作时，上面谈到的这些状态将能够基于检查点一同持久化来保证数据不丢失并且得到可持续的恢复。那么状态在内部是如何表示及存储的呢？这依赖于状态终端的选择。

我们将从几个方面来分解状态终端的实现：

• 被支持的状态类型
• 用户定义的键值对状态
• 状态快照
• 状态访问器
• 状态终端实现

因为状态终端的实现内容较多，所以本文不会太过于拘泥细节，以免管中窥豹

被支持的状态类型

状态相关的接口都维护在package：

org.apache.flink.api.common.state

其继承关系如图：

通过多层的继承，最终的叶子节点是被状态终端直接支持的几种状态类型，它们是：

• ValueState ： 单值状态
• ListState ： 集合状态
• FoldingState ： folding状态，for FoldFunction
• ReducingState ： reducing状态，for ReduceFunction

注意这里只定义了实现这些状态的协议接口，具体的实现本文后面会谈到

针对每一个被直接支持的状态，都有一个描述它们的状态描述符（StateDescriptor），来负责创建对应的状态。一个状态描述符描述状态的名称，默认值。并提供了一个抽象方法来创建状态：

    /**
     * Creates a new {@link State} on the given {@link StateBackend}.
     *
     * @param stateBackend The {@code StateBackend} on which to create the {@link State}.
     */
    public abstract S bind(StateBackend stateBackend) throws Exception;

上面提到的所有被直接支持的状态都有一个描述符：

从上面创建状态的方法bind的签名中可以看到，它依赖于参数StateBackend。而StateBackend暂且可以看作是创建状态的代理。

用户定义的键值对状态

上面的State定义了特定状态的接口协议。除了上面的那些基本状态外，Flink还提供了基于键值对的用户定义的状态，它以KvState接口来描述，其实它才是最终结合检查点机制进行存储和恢复的状态表示。其携带多个泛型参数：

• key的类型
• 命名空间的类型
• 最终存储的State的类型
• 状态描述符StateDescriptor的类型
• 管理该KvState的AbstractStateBackend的具体类型

可以简单地将其看作State的容器

该接口提供了一个snapshot方法，用于结合检查点机制提供快照支持。并返回KvStateSnapshot的实例来表示一个键值对状态的快照。

KvStateSnapshot<K, N, S, SD, Backend> snapshot(long checkpointId, long timestamp) throws Exception;

当然针对每种被直接支持的状态，都有KvState的特定实现：

状态快照

KvStateSnapshot表示KvState快照的接口，它结合检查点机制提供了对状态进行恢复：

• restoreState ： 基于状态终端以及表示检查点的时间戳等来恢复状态

从类图关系可以看出每个针对键值状态的实现（KvState）都有一个内部类提供与之对应的快照实现：

状态访问器

StateHandle给operator提供操作状态的接口，将状态从面向存储介质的原始表示还原为对象表示。重要接口：

T getState(ClassLoader userCodeClassLoader) throws Exception;

可以理解为状态的反序列化接口，根据给定的类加载器加载需要反序列化的类表示来还原状态。

状态终端的实现

所谓的状态终端是真正跟状态持久化介质交互的代理类。

AbstractStateBackend为实现状态终端提供了一个模板。主要提供了如下功能：

• 状态创建/获取、创建快照
• 基于检查点存储状态
• 定义检查点状态输出流

跟检查点有关的部分：

定义了创建状态检查点输出流CheckpointStateOutputView的接口（抽象方法），以及对检查点状态反序列化的接口。这些接口供继承者根据最终的状态终端选择进行实现。

public abstract CheckpointStateOutputStream createCheckpointStateOutputStream(
            long checkpointID, long timestamp) throws Exception;

public abstract <S extends Serializable> StateHandle<S> checkpointStateSerializable(
            S state, long checkpointID, long timestamp) throws Exception;

Flink支持了三种类型的状态终端：

• MemoryStateBackend
• FsStateBackend
• RocksDBStateBackend（第三方开发者实现，本文不进行代码分析）

它们都以AbstractStateBackend为模板：

如果没有进行配置，MemoryStateBackend将是默认的实现。

MemoryStateBackend

MemoryStateBackend在内部将数据以对象的形式保存的Java堆中。键值对状态以及窗口operator以hash table的形式存储值、触发器等。

建立在检查点的机制上，该状态终端将对状态进行快照并且将状态的快照作为检查点应答消息的一部分发送给JobManager（master），JobManager将快照存储在它的堆内存中。

MemoryStateBackend的限制：

• 每个独立状态的大小默认限制在5MB，可以在MemoryStateBackend的构造器中对该值进行增加
• 不管你将状态大小设置得有多大，它都不能大于akka的frame size
• 状态的总占用空间必须适配JobManager的内存空间

推荐在如下场景时使用MemoryStateBackend作为状态终端：

• 本地开发与调试模式
• 只存储很少状态的Job，例如只包含每次只处理一条记录的函数（Map，FlatMap，Filter…）的job

FsStateBackend

FsStateBackend采用文件系统URL(包含type，address，path)的模式进行配置。例如hdfs://namenode:40010/flink/checkpoints或者file:///data/flink/checkpoints。

FsStateBackend将正在处理的数据存储在TaskManager的内存里。结合检查点，它将状态快照写到基于配置的文件系统的文件里。而最小化元数据信息被存储在JobManager的内存里（如果处于高可用模式，元数据将存储在元数据检查点里）。

推荐在如下场景使用FsStateBackend：

• 具有大量状态，很大的窗口，大量键值对状态的Job
• 全程高可用模式

RocksDBStateBackend

RocksDBStateBackend存储正在处理的数据到RocksDB数据库。而RocksDB被存储在TaskManager的数据字典里。结合检查点机制，整个RocksDB数据库将进行快照并被存储到配置的文件系统中。最小化的元数据被存储到JobManager的内存里（如果配置为高可用模式，将会保存到元数据检查点中）。

推荐在如下场景使用RocksDBStateBackend：

• 具有很大的状态，很长的窗口，大量的键值对状态的Job
• 全程高可用状态

注意，使用RocksDBStateBackend时，你能保存的状态仅受到磁盘可用空间的限制。因此，与MemoryStateBackend将状态保存在内存中进行对比，这种状态终端允许你保存非常多的状态。但这也意味着，它所能达到的最大化的吞吐量也将不及MemoryStateBackend。

综合分析

首先来看具体的状态终端对各种状态的实现：

与此对应的KvStateSnapshot也拥有特定的实现：

结合检查点

状态的存储通常是绑定着检查点的，也就是状态会作为检查点的一部分被一同持久化。因此，它具备了fault tolerance的能力。这里我们分成两部分来看：snapshot、restore。

• snapshot

每个最终的状态，都实现了KvState接口（通过间接继承抽象类AbstractHeapState），而实现该接口就必须实现其snapshot方法。这被认为是所有的最终状态都要实现其生产快照的逻辑。当然，这绝大部分逻辑都被AbstractFsState和AbstractMemState给实现了。

具体而言，AbstractFsState利用FsStateBackend创建FsCheckpointStateOutputStream将状态写入检查点对应的路径下（根据检查点编号）。而AbstractMemState则是将其写入到堆内存中（这里甚至都没有用到检查点编号）。

这里有两个状态终端定义的检查点输出流（用于最终的持久化）：

• restore

恢复逻辑分别实现在AbstractFsStateSnapshot和AbstractMemStateSnapshot的restoreState方法中。restoreState的逻辑基本是snapshot的反逻辑，将数据从特定的持久化介质中反序列化回来，并生成KvState对象。

小结

本文梳理了状态终端的实现方式，由于内容较多，因此省略了一些细节实现。但从本文的分析应该基本能理清状态终端如何对状态进行持久化以及恢复。

微信扫码关注公众号：Apache_Flink

QQ扫码关注QQ群：Apache Flink学习交流群（123414680）