State Backends
本文翻译自文档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