本文翻译自文档Streaming Guide / Fault Tolerance / StateBackend

-----------------------------------------------------------------------------------------

使用Data Stream API编写的程序通常以多种形式维护状态：

·  窗口将收集element或在它被触发后聚合element

·  Transformation方法可能会使用key/value状态接口来存储值

·  Transformation方法也可能会实现Checkpointed接口来使其本地变量进入容错机制

相关信息请见Streaming API Guide的文档Working with State。

当检查点机制启动时，上述的状态将在检查点中持久化来应对数据丢失以及恢复。而状态在内部是如何表示的、状态是如何持久化到检查点中以及持久化到哪里都取决于选定的State Backend。

一、可用的State Backends

Flink自带了以下几种开箱即用的state backend：

·  MemoryStateBackend

·  FsStateBackend

·  RocksDBStateBackend

在没有配置的情况下，系统默认使用MemoryStateBackend

1.1 MemoryStateBackend

MemoryStateBackend在内部以Java堆中的对象形式持有数据。Key/Value状态和窗口Operator则持有一个hash表来存储值、trigger等。

在检查点中，该StateBackend将对状态进行快照并将该快照作为检查点接受完成消息(checkpoint acknowledgement message)发送到JobManager(Master)处，在那里它们也是存储在JobManager的Java堆中。

MemoryStateBackend的局限：

·  单个状态的大小默认地被限制到5MB，该限制值可以在MemoryStateBackend的构造函数周增加。

·  不论配置的最大状态大小是多少，状态大小无法大于akka的frame大小（见于Configuration）

·  聚合的状态必须能放入JobManager的内存

MemoryStateBackend适用于以下情景：

·  本地开发以及debug时使用

·  Job只持有很小的状态时，如job只包含那些拥有某时刻数据的方法（Map,FlatMap,Filter…）。此外，Kafka Consumer也只需要很少的状态

1.2 FsStateBackend

FsStateBackend需要使用一个文件系统的URL来配置(type, address, path)，如"hdfs://namenode:40010/flink/checkpoint"或者"file:///data/flink/checkpoints" 。

FsStateBackend在TaskManager的内存中持有in-flight的数据。当进行检查点时，它像状态的快照写入配置好的文件系统及目录下的文件中。而极少的元数据则存储在JobManager的内存中（或者在高可用性模式(high-availability mode)下，存储在元数据检查点中(metadata checkpoint)）。

FsStateBackend适用于以下情景：

·  拥有大状态、长窗口、打key/Value状态的Job。

·  所有高可用性部署中

1.3 RocksDBStateBackend

RocksDBStateBackend使用一个文件系统URL来配置(type, address, path)，例如"hdfs://namenode:40010/flink/checkpoint"或者"file:///data/flink/checkpoints" 。

RocksDBStateBackend在RocksDB数据库中持有in-flight数据，该数据库默认存储在每个TaskManager的数据目录下。当进行检查点时，整个RocksDB数据库将会被检查点到配置的文件系统及目录中去。而极少的元数据则存储在JobManager的内存掣肘（或者在高可用性模式下，存储在元数据检查点中）。

RocksDBStateBackend适用于以下情景：

·  拥有非常大的状态、长窗口、大key/Value状态的Job。

·  所有高可用性部署中。

注意：要使用RocksDBStateBackend，你必须添加正确的maven dependency到你的项目中：

<dependency>

    <groupId>org.apache.flink</groupId>

    <artifactId>flink-statebackend-rocksdb_2.10</artifactId>

    <version>1.2-SNAPSHOT</version>

</dependency>

现在backend不在binary发行版本当中，要将它引入到集群执行中，请见文档Linking with modules not contained in the binary distribution

二、配置一个StateBackend

StateBackend可以每个job单独配置。此外，你可以定义一个默认的StateBackend，它将在Job没有定义一个StateBackend时启用。

2.1 设置单个job的StateBackend

单个job的StateBackend可以在Job的StreamExecutionEnvrionment中设置，代码如下所示：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new FsStateBackend("hdfs://namenode:40010/flink/checkpoints"));

2.2 设置默认StateBackend

默认StateBackend可以在flink-conf.yaml中配置，使用配置关键字state.backend。

配置entry的可能的值为jobmanager(MemoryStateBackend)，filesystem(FsStateBackend)，或者是实现StateBackend工厂接口FsStateBackendFactory的类的完全限定类名(full
qualified class name)

在默认StateBackend设置为filesystem时，配置项state.backend.fs.checkpointdir定义了检查点数据存储的目录。

配置文件实例如下所示：

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

state.backend: filesystem

# Directory for storing checkpoints

state.backend.fs.checkpointdir: hdfs://namenode:40010/flink/checkpoints