Flink DataStream State Backend（状态后端）

古城微笑少年丶 2021-12-18 23:19 516阅读 0赞

## Flink DataStream State Backend（状态后端） ##

### State Backend（状态后端） ###

用`Data Stream Api` 编写的程序经常是需要保存状态的，如下面情况：

*  `Windows`中聚集的元素直到被触发计算。在触发之前，需要保存`Windows`中的元素的状态。
 *  转换（`Transformation`）函数可以使用`Key/Value`状态接口存储状态值
 *  转换（`Transformation`）函数可以实现`Checkpointed Function`接口使局部变量具有容错能力

激活检查点时，检查点会持续保持此状态，以防止数据丢失并始终如一地恢复。State如何在内部表示，以及在检查点上如何保持以及在何处保持**状态**取决于所选择的**State Backend**。

#### 一、State Backend（状态后端）的种类 ####

当`checkpoint`可用时，State是被持续储存到checkpoint，以防数据丢失，同时可以持续的复原。State储存在哪里，是依赖与`State Backend`的实现。`State Backend`有3种，分别为：

*  **MemoryStateBackend**: State存储在Memory（内存）中。
    
    `MemoryStateBackend`将State作为Java对象保存（在堆内存），存储着key/value状态、window运算符、触发器等的哈希表。
    
    在Checkpoint时，`State Backend`将对State进行快照，并将其作为checkpoint发送到`JobManager`机器上，`JobManager`将这个State数据存储在Java堆内存。
    
    `MemoryStateBackend`默认使用异步快照，来避免阻塞管道。如果需要修改，可以在`MemoryStateBackend`的构造函数将布尔值改为`false`（仅用于调试）。
    
        new MemoryStateBackend(MAX_MEM_STATE_SIZE, false); // false：禁用异步快照
    
    **MemoryStateBackend的局限**：
    
     *  默认情况下，每个State大小限制不超过5MB，可以在`MemoryStateBackend`构造函数增加值
     *  无论最大的State大小如何配置，但都不能超过**Akka的传输最大消息大小（10MB）**
     *  State状态存储在`JobManager`的内存当中
    
    **MemoryStateBackend一般用于本地开发和调试**。
 *  **FsStateBackend**: 使用可靠地文件存储系统State，如HDFS。
    
    FsStateBackend将正在运行的数据保存在`TaskManager`的内存中。在checkpoint时，它将State的快照写入文件系统对应的目录下的文件中。最小元数据存储在`JobManager`的内存中（如果是高可用模式下，元数据存储在checkpoint中）。
    
    FsStateBackend默认使用异步快照，来避免阻塞处理的管道。如果需要禁用，在FsStateBackend构造方法中将布尔值设为false。如：
    
        new FsStateBackend(path,flase);// path:checkpoint的fs路径，false：禁用异步快照
    
    FsStateBackend适用于**具有大状态、长窗口大键值的高可用作业**。
 *  **RocksDBStateBackend**: 使用RocksDB存储State。
    
    `RocksDBStateBackend`将正在运行的数据保存在RocksDB数据库中。该数据库（默认情况下）存储在`TaskManager`数据目录中。在checkpoint时，整个RocksDB数据库将被checkpoint带配置的文件系统对应的目录下。最小元数据㽾在`JobManager`的内存中（高可用模式下，元数据储存在checkpoint目录下）
    
    **RocksDBStateBackend始终执行异步快照**。
    
    RocksDBStateBackend的局限性：
    
     *  由于RocksDB的JNI网桥API基于byte \[\]，因此每个密钥和每个值支持的最大大小为2 ^ 31个字节。
    
    RocksDBStateBackend适用于非常大的状态，长窗口、大键值状态的高可用作业。
    
    **RocksDBStateBackend是目前唯一提供增量checkpoint的后端**。
    
    #### 二、配置State Backend（状态后端） ####
    
    如果不指定`State Backend`，则默认`State Backend`是`JobManager`。如果要修改为集群默认的`State Backend`实现，可以通过在`flink-conf.yaml`中定义新的`State Backend`来实现。
    
    在`flink-conf.yaml`配置文件中指定`state.backend`参数的值。该参数的值有：`jobmanager(MemoryStateBackend)`，`filesystem(FsStateBackend)`，`rocksdb(RocksDBStateBackend)`。
    
    eg:
    
        # 设置为FsStateBackend
        state.backend: filesystem
        
        # 配置文件储存路径
        state.checkpoints.dir: hdfs://namenode:40010/flink/checkpoint
    
    如果不想集群配置的State Backend满足不了，我们也可以通过在代码层面进行修改，如：
    
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置FsStateBackend为状态后端
        env.setStateBackend(new FsStateBackend("hdfs://namenode:40010/flink/checkpoints"));
        
        // 或者设置RocksDBStateBackend为状态后端
        // env.setStateBackend(new RocksDBStateBackend("hdfs://namenode:40010/flink/checkpoints"));
    
    如果使用RocksDBStateBackend，则必须加入依赖：
    
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
            <version>1.8.0</version>
        </dependency>
    
    **RocksDB State Backend配置选项**
    
     *  `state.backend.rocksdb.checkpoint.transfer.thread.num`: 用于在RocksDBStateBackend中传输（上传和下载）文件的线程数。默认1。
     *  `state.backend.rocksdb.localdir`：RocksDB放置在local下的目录（TaskManager机器目录）。没有默认值。
     *  `state.backend.rocksdb.options-factory`: 工厂类，用于创建DBOptions和ColumnFamilyOptions。默认为`org.apache.flink.contrib.streaming.state.DefaultConfigurableOptionsFactory`，它将读取`RocksDBConfigurableOptions`中提供的已配置选项。
     *  `state.backend.rocksdb.predefined-options`: RocksDB的DBOptions和ColumnFamilyOptions的预定义设置。
        
        当前支持的候选预定义选项是：
        
         *  DEFAULT（默认）
         *  SPINNING\_DISK\_OPTIMIZED
         *  SPINNING\_DISK\_OPTIMIZED\_HIGH\_MEM
         *  FLASH\_SSD\_OPTIMIZED
        
        请注意，OptionsFactory中的用户自定义选项和选项应用于这些预定义的选项和选项之上
     *  `state.backend.rocksdb.timer-service.factory`：工厂类，计时器服务状态实现。选项是`HEAP`或`ROCKSDB`，默认`HEAP`。
     *  `state.backend.rocksdb.ttl.compaction.filter.enabled`：是否启用TTL的清理状态的压缩过滤器。默认false。