3.6、Flink流处理（Stream API）- State & Fault Tolerance（状态和容错）之 State Backends（状态后端）

以你之姓@ 2023-06-29 02:53 24阅读 0赞

Flink提供了不同的状态后端，用于指定状态存储的方式和位置。

状态可以位于Java的堆上，也可以位于堆外。根据您的状态后端，Flink 也可以管理应用程序的状态，这意味着 Flink 处理内存管理(如果必要的话可能会溢出到磁盘)，以允许应用程序保存非常大的状态。默认情况下，配置文件是 flink-conf.yaml 管理所有 Flink 作业的状态后端。也可以根据每个作业重写默认状态后端，如下所示。

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

在数据流API中编写的程序通常以各种形式保持状态：

*  窗口收集元素或聚合，直到它们被触发
 *  转换函数可以使用键/值状态接口来存储值
 *  转换函数可以实现 CheckpointedFunction 接口，使其局部变量具有容错能力

当检查点被激活时，这种状态在检查点上被持久化，以防止数据丢失和一致地恢复。如何在内部表示状态，以及如何在检查点上持久化状态和持久化位置取决于所选的状态后端。

## 可用状态后端 ##

开箱即用，Flink 包这些状态后端：

*  *MemoryStateBackend*
 *  *FsStateBackend*
 *  *RocksDBStateBackend*

> 如果没有其他配置，系统将使用 memorystateback。

### MemoryStateBackend ###

memorystateback 在内部将数据作为对象保存在 Java 堆中。键/值状态和窗口算子持有存储值、触发器等的哈希表。

在检查点上，此状态后端将对状态进行快照，并将其作为检查点确认消息的一部分发送给 JobManager(主)，后者也将其存储在堆中。

可以将memorystateback配置为使用异步快照。虽然我们强烈建议使用异步快照来避免阻塞管道，但请注意，这是目前默认启用的。要禁用此功能，用户可以实例化一个memorystateback，并在构造函数中将相应的布尔标志设置为false(这应只用于调试)，例如：

new MemoryStateBackend(MAX_MEM_STATE_SIZE, false);

memorystateback的限制：

*  默认情况下，每个状态的大小限制为5 MB。这个值可以在memorystateback的构造函数中增加。
 *  不管配置的最大状态大小如何，状态都不能大于 akka frame 大小（akka.framesize）(参见配置)。
 *  聚合状态必须适合于JobManager内存。

建议使用 memorystateback 的情况：

*  本地开发与调试
 *  保存很少状态的作业，例如只包含一次记录功能的作业(Map、FlatMap、Filter等)。Kafka 消费者只需要很少的状态。

### FsStateBackend ###

fsstatebackend 使用文件系统 URL(类型、地址、路径)进行配置，如“hdfs://namenode:40010/flink/checkpoints”或“file:///data/flink/checkpoints”。

fsstatebackend 在 taskmanager 的内存中保存流中的数据。在检查点时，它将状态快照写入配置的文件系统和目录中的文件。最小的元数据存储在 JobManager 的内存中(或者，在高可用性模式下，存储在元数据检查点中)。

fsstatebackend 默认使用异步快照，以避免在编写状态检查点时阻塞处理管道。要禁用此功能，用户可以实例化一个 fsstatebackend，并在构造函数中将相应的布尔标志设置为false，例如：

new FsStateBackend(path, false);

fsstatebackend 使用建议：

*  具有大状态、长窗口、大键/值状态的作业。
 *  所有高可用性设置。

### RocksDBStateBackend ###

rocksdbstateback 使用文件系统URL(类型、地址、路径)进行配置，例如“hdfs://namenode:40010/flink/检查点”或“file:///data/flink/检查点”。

rocksdbstateback 后端在 RocksDB 数据库中保存动态数据，该数据库(默认情况下)存储在 TaskManager 数据目录中。在检查点时，整个 RocksDB 数据库将存储到配置的文件系统和目录中。最小的元数据存储在 JobManager 的内存中(或者，在高可用性模式下，存储在元数据检查点中)。

> rocksdbstateback 后端总是执行异步快照。

**局限性：**

RocksDB’s JNI bridge API 是基于 byte\[\] 的，所以每个键和每个值所支持的最大大小是2^31字节。**重要提示：**在 RocksDB 中使用合并操作的状态(例如 ListState)可以悄悄地累积值大小> 2^31字节，然后在下一次检索时失败。这是 RocksDB JNI 当前的一个限制。

**使用建议：**

*  具有非常大的状态、长窗口、大键/值状态的作业。
 *  所有高可用性设置。

注意：存储状态的大小受可用磁盘的大小的限制，与保存在内存的 fsstatebacnend 相比，可以存储更大的数据。然而，这也意味着使用此状态后端可以实现的最大吞吐量会更低。所有的读写操作都必须经过序列化和反序列话来检索存储状态对象，意味着消耗的资源和时间比基于内存的要多。

rocksdbstateback 是目前唯一提供增量检查点的后端([3.4、Flink 集群部署（Deployment & Operations）- State & Fault Tolerance（Flink状态和容错）之 调优检查点和大状态][3.4_Flink _Deployment _ Operations_- State _ Fault Tolerance_Flink_])。

某些 RocksDB 原生指标是可用的，但在默认情况下是禁用的，详见：[Flink 集群部署（Deployment & Operations）- Configuration（Flink配置项）][Flink _Deployment _ Operations_- Configuration_Flink]

## 状态后端配置 ##

如果没有指定任何内容，则默认的状态后端是 jobmanager。如果希望为集群上的所有作业建立不同的缺省值，可以通过在 flink- config.yaml 中定义新的缺省状态后端来实现。可以根据每个作业重写默认状态后端，如下所示。

### Per-job State Backend ###

per-job state backend 在 `StreamExecutionEnvironment 上设置，例如:`

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

如果希望在IDE中使用 rocksdbstateback 或在 Flink 作业中以编程方式配置它，则必须将以下依赖项添加到Flink项目中。

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
        <version>1.9.0</version>
        <scope>provided</scope>
    </dependency>

> 因为 RocksDB 是默认的 Flink 发行版的一部分，如果您在作业中没有使用任何 RocksDB 代码，并且通过state配置状态后端，那么就不需要这个依赖项。在您的 flink- confi .yaml 中提供后端和进一步的检查点以及特定于 rocksdb 的参数。

### Default State Backend ###

默认的状态后端可以通过 flink-config.yaml 中的 state.bacnend 配置。

配置项的可能值是 jobmanager (memorystatebackend)、filesystem (fsstatebackend)、rocksdb (rocksdbstatebackend)或实现 StateBackendFactory 的类的完全合格类名，如 org.apache.flink.contrib.stream .state.RocksDBStateBackendFactory RocksDBStateBackend。

`state.checkpoints.dir `选项定义所有后端写入检查点数据和元数据文件的目录。

配置文件中的示例部分如下所示：

# The backend that will be used to store operator state checkpoints
    
    state.backend: filesystem
    
    
    # Directory for storing checkpoints
    
    state.checkpoints.dir: hdfs://namenode:40010/flink/checkpoints

**RocksDB State Backend Config Options**

<table> 
 <thead> 
  <tr> 
   <th>Key</th> 
   <th style="width:224px;">Default</th> 
   <th style="width:563px;">Description</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td> <p>state.backend.rocksdb.checkpoint.transfer.thread.num</p> </td> 
   <td style="width:224px;">1</td> 
   <td style="width:563px;">rocksdbstateback 中用于传输(下载和上传)文件的线程数(每个有状态算子)。</td> 
  </tr> 
  <tr> 
   <td> <p>state.backend.rocksdb.localdir</p> </td> 
   <td style="width:224px;">(none)</td> 
   <td style="width:563px;">本地目录(在TaskManager上)，RocksDB 将其文件放在这里。</td> 
  </tr> 
  <tr> 
   <td> <p>state.backend.rocksdb.options-factory</p> </td> 
   <td style="width:224px;">"org.apache.flink.contrib.streaming.state.DefaultConfigurableOptionsFactory"</td> 
   <td style="width:563px;">用于 RocksDB 创建 DBOptions 和 ColumnFamilyOptions 的选项工厂类。 默认的选项工厂是org.apache.flink.contrib.stream .state。DefaultConfigurableOptionsFactory，它会读取 'RocksDBConfigurableOptions' 中提供的配置选项。</td> 
  </tr> 
  <tr> 
   <td> <p>state.backend.rocksdb.predefined-options</p> </td> 
   <td style="width:224px;">"DEFAULT"</td> 
   <td style="width:563px;">Flink community 为 RocksDB DBOptions 和 ColumnFamilyOptions 预定义的设置。 当前支持的候选预定义选项默认为 DEFAULT、SPINNING_DISK_OPTIMIZED、SPINNING_DISK_OPTIMIZED_HIGH_MEM 、FLASH_SSD_OPTIMIZED. 注意，用户自定义选项和来自OptionsFactory的选项应用于这些预定义的选项之上。</td> 
  </tr> 
  <tr> 
   <td> <p>state.backend.rocksdb.timer-service.factory</p> </td> 
   <td style="width:224px;">"HEAP"</td> 
   <td style="width:563px;">这决定了定时器服务状态实现的工厂。对于基于 ROCKSDB 的实现，可以选择堆(基于堆，默认)或 ROCKSDB。</td> 
  </tr> 
  <tr> 
   <td> <p>state.backend.rocksdb.ttl.compaction.filter.enabled</p> </td> 
   <td style="width:224px;">false</td> 
   <td style="width:563px;">这将确定后端是否启用了TTL的压缩过滤器以清除状态。<strong>注意：</strong>用户仍然可以在状态描述符的状态 TTL 配置中决定过滤器对于特定的状态是否激活。</td> 
  </tr> 
 </tbody> 
</table>

原文地址：

[https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/stream/state/state\_backends.html][https_ci.apache.org_projects_flink_flink-docs-release-1.9_dev_stream_state_state_backends.html]

[https://ci.apache.org/projects/flink/flink-docs-release-1.9/ops/state/state\_backends.html][https_ci.apache.org_projects_flink_flink-docs-release-1.9_ops_state_state_backends.html]

[3.4_Flink _Deployment _ Operations_- State _ Fault Tolerance_Flink_]: https://blog.csdn.net/zpf_940810653842/article/details/103957084
[Flink _Deployment _ Operations_- Configuration_Flink]: https://blog.csdn.net/zpf_940810653842/article/details/99873275
[https_ci.apache.org_projects_flink_flink-docs-release-1.9_dev_stream_state_state_backends.html]: https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/stream/state/state_backends.html
[https_ci.apache.org_projects_flink_flink-docs-release-1.9_ops_state_state_backends.html]: https://ci.apache.org/projects/flink/flink-docs-release-1.9/ops/state/state_backends.html