【Flink-State】Flink三种StateBackend的区别与使用：MemoryStateBackend & FsStateBackend & RocksDBStateBackend

末蓝、 2023-10-08 19:12 34阅读 0赞

#### 【Flink-State】Flink三种StateBackend的区别与使用：MemoryStateBackend & FsStateBackend & RocksDBStateBackend ####

*  1）场景描述
 *  2）StateBackend的存储位置与配置方式
 *  3）StateBackend的种类
 *  4）StateBackend的介绍
 *   *  4.1.MemoryStateBackend
     *  4.2.FsStateBackend
     *  4.3.RocksDBStateBackend
 *  5）StateBackend的使用
 *   *  5.1.不同StateBackend吞吐量对比
     *  5.2.不同StateBackend延迟对比
     *  5.3.StateBackend的选择
     *  5.4.StateBackend的设置方式

## 1）场景描述 ##

当Flink程序的`checkpoint`被触发时，状态会被持久化到`checkpoint`，以防止数据丢失和无缝恢复。

状态在内部如何组织和它们如何以及在哪持久化，依赖于所选的状态后端。

## 2）StateBackend的存储位置与配置方式 ##

![在这里插入图片描述][a18ea12650cf4c048ac89c45df50d6b6.png]

**（1）StateBackend的存储位置**

`Checkpoint`的存储的位置取决于配置的`StateBackend`（`JobManager 内存，文件系统，数据库...`）

默认情况下，`State`存储在`TaskManager内存`中，`Checkpoint`存储在 `JobManager内存`中。

**（2）StateBackend的配置方式**

Flink 支持在其他StateBackend中存储 State 和 Checkpoint。可以通过如下方法配置：`StreamExecutionEnvironment.setStateBackend(…)`

Flink 提供了不同的StateBackend，支持不同的State存储方式和位置。默认会使用配置文件`flink-conf.yaml`指定的选项，也可以在每个作业中设置来覆盖默认选项：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setStateBackend(...);

## 3）StateBackend的种类 ##

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

*  MemoryStateBackend
 *  FsStateBackend
 *  RocksDBStateBackend

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

## 4）StateBackend的介绍 ##

`MemoryStateBackend`：默认，小状态，本地调试使用。

`FsStateBackend`：大状态，长窗口，高可用场景。

`RocksDBStateBackend`：超大状态，长窗口，高可用场景，可增量checkpoint。

### 4.1.MemoryStateBackend ###

**（1）了解MemoryStateBackend**

1、MemoryStateBackend将内部的数据保存在Java堆上。Key/value状态和窗口操作符持有存储值，触发器等的哈希表。

2、当进行checkpoint时，这个状态后端会对当前的状态进行快照，并且将其作为checkpoint ACK消息的一部分发送给JobManager(master)，该JobManager将其存储在它的堆上。

3、MemoryStateBackend的四个构造函数

public MemoryStateBackend() {
        	
            this(5242880);	
        }	
    public MemoryStateBackend(int maxStateSize) {
        	
        this(maxStateSize, true);	
    }	
    public MemoryStateBackend(boolean asynchronousSnapshots) {
        	
        this(5242880, asynchronousSnapshots);	
    }	
    public MemoryStateBackend(int maxStateSize, boolean asynchronousSnapshots) {
        	
        this.maxStateSize = maxStateSize;	
        this.asynchronousSnapshots = asynchronousSnapshots;	
    }

注意：

MemoryStateBackend可以配置使用异步快照的方式。

虽然我们强烈鼓励使用异步快照的方式来避免管道阻塞，但是请注意，这个是一个新特性，目前默认情况下不启用。

为了启用这个状态，用户可以在初始化MemoryStateBackend时将构造函数中相应的布尔标识设为true，例如：

new MemoryStateBackend(MAX_MEM_STATE_SIZE, true);

**（2）MemoryStateBackend的局限性**

*  单个状态的大小默认情况下最大为5MB。这个值可以通过MemoryStateBackend构造函数进行增加。
 *  无论配置的最大状态大小为多少，状态的大小不能超过akka帧大小。
 *  聚合的状态必须在JobManager的内存中能存放。

**（3）MemoryStateBackend适用于**

*  本地开发和调试。
 *  只有很小状态的作业，例如作业只由record-at-a-time函数组成(Map，FlatMap，Filter，…)。Kafka消费者只需要非常小的状态。

### 4.2.FsStateBackend ###

**（1）了解FsStateBackend**

1、FsStateBackend默认使用异步快照，对每个快照文件大小有要求：\[0, 1048576\]，且状态快照大小不能超过TaskManager的内存。但状态快照最终保存在文件系统中，所以FsStateBackend适用于大数据的生产环境，可处理长窗口，大状态或大key-value状态任务。

2、FsStateBackend将流计算数据状态存储在<font color=orangeTaskManager的内存中，在数据流遇到检查点屏障时，再将数据快照存储在配置好的文件系统中，在<font color=orangeJobManager内存中会存储少量的checkpoint元数据。

3、FsStateBackend的三种构造函数：

public FsStateBackend(String checkpointDataUri, boolean asynchronousSnapshots) throws IOException {
        	
        this(new Path(checkpointDataUri), asynchronousSnapshots);	
    }	
     
    	
    public FsStateBackend(Path checkpointDataUri, boolean asynchronousSnapshots) throws IOException {
        	
        this(checkpointDataUri.toUri(), asynchronousSnapshots);	
    }	
     
    	
    //fileStateSizeThreshold默认1024	
    public FsStateBackend(URI checkpointDataUri, int fileStateSizeThreshold, boolean asynchronousSnapshots) throws IOException {
        	
        Preconditions.checkArgument(fileStateSizeThreshold >= 0, "The threshold for file state size must be zero or larger.");	
        Preconditions.checkArgument(fileStateSizeThreshold <= 1048576, "The threshold for file state size cannot be larger than %s", new Object[]{
        1048576});	
        this.fileStateThreshold = fileStateSizeThreshold;	
        this.basePath = validateAndNormalizeUri(checkpointDataUri);	
        this.asynchronousSnapshots = asynchronousSnapshots;	
    }

**（2）FsStateBackend的存储路径**

*  FsStateBackend需要配置存储的文件系统，可以是`HDFS`路径

hdfs://namenode:40010/flink/checkpoints

*  也可以是`文件系统`路径

file:///data/flink/checkpoints

**（3）FsStateBackend的推荐使用的场景**

*  处理大状态，长窗口，或大键值状态的有状态处理任务， 例如分钟级窗口聚合或 join。
 *  适合用于高可用方案（需要开启HA的作业）
 *  可以在生产环境中使用。

### 4.3.RocksDBStateBackend ###

**（1）了解RocksDBStateBackend**

1、RocksDBStateBackend的存储位置

RocksDBStateBackend将in-flight数据存储在RocksDB数据库中，它(默认)存储在TaskManager的data目录下。

当checkpoint时，整个RocksDB数据库将被checkpoint到配置的文件系统和目录下。最小的元数据存储在JobManager的内存中(或者，在高可用模式下，在元数据checkpoint中)

2、RocksDB介绍

RocksDB是一种嵌入式的本地数据库。RocksDBStateBackend 将处理中的数据使用RocksDB存储在本地磁盘上。

在checkpoint时，整个RocksDB数据库会被存储到配置的文件系统中，或者在超大状态作业时可以将增量的数据存储到配置的文件系统中。同时Flink会将极少的元数据存储在JobManager 的内存中，或者在Zookeeper中（对于高可用的情况）

RocksDB默认也是配置成异步快照的模式。RocksDB是一个 key/value 的内存存储系统，和其他的 key/value 一样，先将状态放到内存中，如果内存快满时，则写入到磁盘中，但需要注意RocksDB不支持同步的Checkpoint，构造方法中没有同步快照这个选项。不过\`RocksDB支持增量的 Checkpoint，意味着并不需要把所有sst文件上传到 Checkpoint目录，仅需要上传新生成的sst文件即可。

它的Checkpoint存储在外部文件系统（本地或HDFS），其容量限制只要单个 TaskManager上State总量不超过它的内存+磁盘，单Key最大2G，总大小不超过配置的文件系统容量即可。

**（2）RocksDBStateBackend局限性**

*  RocksDB支持的单key和单value的大小最大为每个2^31字节。这是因为 RocksDB 的JNI API是基于byte\[\]的。
 *  对于使用具有合并操作的状态的应用程序，例如 ListState，随着时间可能会累积到超过2^31字节大小，这将会导致在接下来的查询中失败。

**（3）RocksDBStateBackend推荐使用的场景**

*  最适合用于处理大状态，长窗口，或大键值状态的有状态处理任务。
 *  非常适合用于高可用方案。
 *  最好是对状态读写性能要求不高的作业。

注意：

RocksDBStateBackend是目前唯一提供增量checkpoint的状态后端。

## 5）StateBackend的使用 ##

### 5.1.不同StateBackend吞吐量对比 ###

![在这里插入图片描述][bad5f033cf4d47f5b47dcae1c0b5391e.png]

结论：

**（1）使用FileSystem和Memory的吞吐差异不大（都是使用堆内存管理处理中的数据），使用RocksDB的吞吐差距明显。**

**（2）Standalone和on Yarn的总体差异不大，使用FileSystem和Memory时 on Yarn模式下吞吐稍高，相反的使用RocksDB时Standalone模式下的吞吐稍高。**

### 5.2.不同StateBackend延迟对比 ###

![在这里插入图片描述][cd48cc95ce7743ffa8ed878248e26278.png]

结论：

**（1）使用FileSystem和Memory时延迟基本一致且较低。**

**（2）使用RocksDB时延迟稍高，且由于吞吐较低，在达到吞吐瓶颈附近延迟陡增。其中on Yarn模式下吞吐更低，延迟变化更加明显。**

### 5.3.StateBackend的选择 ###

<table> 
 <thead> 
  <tr> 
   <th align="center"><strong>StateBackend</strong></th> 
   <th align="center"><strong>in-flight</strong></th> 
   <th align="center"><strong>checkpoint</strong></th> 
   <th align="center">吞吐</th> 
   <th align="center"><strong>推荐使用场景</strong></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center">MemoryStateBackend</td> 
   <td align="center">TM Memory</td> 
   <td align="center">JM Memory</td> 
   <td align="center">高</td> 
   <td align="center">调试、无状态或对数据丢失或重复无要求</td> 
  </tr> 
  <tr> 
   <td align="center">FsStateBackend</td> 
   <td align="center">TM Memory</td> 
   <td align="center">FS/HDFS</td> 
   <td align="center">高</td> 
   <td align="center">普通状态、窗口、KV 结构</td> 
  </tr> 
  <tr> 
   <td align="center">RocksDBStateBackend</td> 
   <td align="center">RocksDB on TM</td> 
   <td align="center">FS/HDFS</td> 
   <td align="center">低</td> 
   <td align="center">超大状态、超长窗口、大型 KV 结构</td> 
  </tr> 
 </tbody> 
</table>

注意：

（1）如果不做任何指定，默认的状态后端是`JobManager`。

（2）如果你希望为你的集群中的所有作业创建一个非默认的状态后端，你可以通过在`flink-conf.yaml`中指定一个新的默认后端。默认的状态后端可以在每个作业的基础上进行覆盖。

### 5.4.StateBackend的设置方式 ###

**（1）作业的状态后端通过作业中的`StreamExecutionEnvironment`进行设置**

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

**（2）默认状态后端可以通过在`flink-conf.yaml`中设置state.backend值指定**

1、配置项

*  jobmanager (MemoryStateBackend)
 *  filesystem (FsStateBackend)
 *  rocksdb (RocksDBStateBackend)
 *  或者实现了状态后端工厂`FsStateBackendFactory`的类的完全限定类名，例如，为`RocksDBStateBackend`设置为`org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory`

2、配置文件示例

# 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

[a18ea12650cf4c048ac89c45df50d6b6.png]: https://img-blog.csdnimg.cn/a18ea12650cf4c048ac89c45df50d6b6.png
[bad5f033cf4d47f5b47dcae1c0b5391e.png]: https://img-blog.csdnimg.cn/bad5f033cf4d47f5b47dcae1c0b5391e.png
[cd48cc95ce7743ffa8ed878248e26278.png]: https://img-blog.csdnimg.cn/cd48cc95ce7743ffa8ed878248e26278.png