Flink DataStream State(状态)

「爱情、让人受尽委屈。」 2021-12-18 23:21 427阅读 0赞

## Flink DataStream State(状态) ##

#### 一、State（状态）的种类 ####

State有2类：**Keyed State**和**Operator State**。

*  **Keyed State**：与keys相关的状态，仅仅用于Keyed Stream的Operator(算子)的Function。
 *  **Operator State**: Non-Keyed State，每个operator state都绑定到一个并行度算子实例。
    
    当并行度更改时，运算符状态接口支持在并行运算符实例之间重新分配状态。进行这种重新分配可能有不同的Schema。

**Keyed State和Operator State都存在2种形式**：

*  **Raw State**：是指算子（Operator）保持他们自己数据的结构。当checkpoint时，他们仅仅写入一些列的byte字节。Flink不知道这个状态的数据结构，看到的仅仅是原始的字节bytes。
 *  **Managed State**：代表着Flink在运行时期所控制的数据结构。例如内部哈希表，RocksDB。Managed State对象有：`ValueState、ListState、MapState、ReducingState`等。Flink在运行时期对状态进行编码，在checkpoint是写入。

推荐使用**Managed State（托管状态）**。当程序的并行度发生改变，Flink可以自动重新分发状态。同时`Managed State(托管状态)`在内存中管理的更好。

#### 二、State的生存时间（TTL） ####

如果需要使用State生存时间，你需要在创建Managed State是通过StateTtlConfig对象配置时间。

例如：示例代码

...
    StateTtlConfig ttlConfig = StateTtlConfig
        .newBuilder(Time.seconds(1))
        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
        .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
        .build();
        
    ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("text state", String.class);
    stateDescriptor.enableTimeToLive(ttlConfig);
    ...

该`newBuilder`方法的第一个参数是必需的，它是生存时间值。

`setUpdateType()`：设置State更新类型，配置参数有：

*  `StateTtlConfig.UpdateType.Disabled`：状态不过期
 *  `StateTtlConfig.UpdateType.OnCreateAndWrite`：仅仅是创建和写入权限（默认）
 *  `StateTtlConfig.UpdateType.OnReadAndWrite`：读和写权限

`setStateVisibility()`：State可见状态，配置参数有：

*  `StateTtlConfig.StateVisibility.NeverReturnExpired`：过期值永不返回
 *  `StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp`：如果可用，则返回（没清除，还存在，返回）

注意：

1.使用TTL，会储存上次修改时间戳和用户值，增加状态存储的消耗。

2.**目前仅支持Processing Time的TTL**。

3.TTL不是checkpoint和savepoint的一部分。

4.尝试使用启用TTL的描述符恢复以前配置为不带TTL的状态，或者使用启用TTL的描述符恢复之前配置的状态，将导致兼容性失败和StateMigrationException。

#### 三、State的清除策略 ####

State的清除策略有3种，分别是：**EMPTY\_STRATEGY**（空策略，不清除），**FULL\_STATE\_SCAN\_SNAPSHOT**（清除完成快照）、**INCREMENTAL\_CLEANUP**（增量清除）、**ROCKSDB\_COMPACTION\_FILTER**（RocksDB压缩过滤清除）。

其中`INCREMENTAL_CLEANUP`和`ROCKSDB_COMPACTION_FILTER`策略是在后台清除。

*  **FULL\_STATE\_SCAN\_SNAPSHOT**：清除完成快照，在获取完整状态快照时激活清理，这将减小其大小。
    
    示例代码：
    
        StateTtlConfig ttlConfig = StateTtlConfig
            .newBuilder(Time.minutes(1))
            .cleanupFullSnapshot() // 清除完整快照
            .build();
    
    这个`cleanupFullSnapshot()`方法，此选项不适用于`RocksDB`状态后端中的增量检查点。在调用`cleanupFullSnapshot()`方法，表示激活`FULL_STATE_SCAN_SNAPSHOT`策略。
    
    对于存在的jobs，可以随时在`statettlconfig`中激活或停用此清除策略。例如从`savepoint`重新启动后。

**3.2 增量清除**

*  ##### INCREMENTAL\_CLEANUP：增量清理 #####
    
    对应的清除策略是：`INCREMENTAL_CLEANUP`，调用`cleanupIncrementally()`方法，激活此策略。
    
    代码示例：
    
        StateTtlConfig ttlConfig = StateTtlConfig
            .newBuilder(Time.minutes(1))
            .cleanupIncrementally(100, false)
            .build();
    
    `cleanupIncrementally(int cleanupSize, boolean runCleanupForEveryRecord )`：
    
     参数1：每次触发清除状态的最大key数
    
     参数2：是否对每个已处理的记录运行增量清理。
    
    **目前这个增量清理，只支持Heap State（JobManager内存），如果是RocksDB将不生效**。
    
    *如果将堆状态后端与**同步快照**一起使用，则全局迭代器将保留所有键的副本。因为它的具体实现不支持并发修改而进行迭代。启用此功能将增加内存消耗。异步快照没有此问题*。
    
    如果默认后台清理，则将为Heap State激活此策略。其中对于每条记录，有5项检查不清除：
    
     *  如果状态没有访问或没有处理记录，则过期状态将持续存在。
     *  增量清理所花费的时间会增加记录处理延迟。
     *  目前，仅针对Heap状态后端实施增量清理。为RocksDB设置它将不起作用。
     *  如果堆状态后端与同步快照一起使用，则全局迭代器会在迭代时保留所有键的副本，因为它的特定实现不支持并发修改。启用此功能会增加内存消耗。异步快照没有此问题。
     *  对于现有作业，可以随时激活或停用此清理策略StateTtlConfig，例如在从保存点重新启动之后。
 *  ##### ROCKSDB\_COMPACTION\_FILTER: RocksDB压缩过滤清理 #####
    
    清除策略是：**ROCKSDB\_COMPACTION\_FILTER**。在调用`cleanupInRocksdbCompactFilter()`方法时，激活策略。
    
    RocksDB定期运行异步压缩以merge状态更新并减少存储。Flink压缩过滤器使用TTL检查状态条目的到期时间戳，并排除过期值。
    
    默认情况下禁用此功能。必须先通过设置Flink配置选项`state.backend.rocksdb.ttl.compaction.filter.enabled`设置为`true`或通Api调用`RocksDBStateBackend::enableTtlCompactionFilter`是否为作业创建自定义RocksDB状态后端来激活RocksDB后端。然后可以将任何具有TTL的状态配置为使用过滤器。
    
    `cleanupInRocksdbCompactFilter()`示例
    
        StateTtlConfig ttlConfig = StateTtlConfig
            .newBuilder(Time.seconds(1))
            .cleanupInRocksdbCompactFilter(1000L) // 1000：表示在更新当前时间戳之前由压缩筛选器处理的状态条目数。默认1000L。
            .build();
    
     *  在压缩期间调用TTL过滤器会减慢它的速度。TTL过滤器必须解析上次访问的时间戳，并检查每个正在压缩的密钥的每个存储状态条目的到期时间。在收集状态类型（列表或映射）的情况下，还对每个存储的元素调用检查。
     *  如果此功能与包含非固定字节长度的元素的列表状态一起使用，则本机TTL过滤器必须每个状态条目另外调用JNI上元素的Flink Java类型序列化程序，其中至少第一个元素已到期确定下一个未到期元素的偏移量。
     *  对于现有作业，可以随时激活或停用此清理策略StateTtlConfig，例如在从保存点重新启动之后