Flink Program Guide (9) -- StateBackend : Fault Tolerance(Basic API Concepts -- For Java)

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" 。

RocksDBStateBackendRocksDB数据库中持有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

  

上一篇:(转载)Cocos2dx-OpenGL ES 2.0教程:你的第一个三角形(1)


下一篇:异构数据库迁移 db2---oracle