Flink 状态管理详解（State TTL、Operator state、Keyed state）

kk大数据

发布于 2021-02-24 15:21:12

6.7K0

发布于 2021-02-24 15:21:12

一、简介

Flink官网的自我介绍：Apache Flink® — Stateful Computations over Data Streams，可以看出状态计算是 Flink 引以为豪的杀手锏。那什么是带状态的计算呢？简单说计算任务的结果不仅仅依赖于输入，还依赖于它的当前状态。

实时计算如果任务失败导致中间状态丢失，将是一个非常可怕的事情。

比如实时计算每天的 pv,uv 等指标，任务掉线后中间状态也丢失了，那只能从凌晨数据重新计算。

如果是有状态的计算大可不必担心，从任务掉线的时刻继续计算，妈妈再也不用担心我的任务掉线了。

下面介绍一下Flink如何实现状态计算和状态管理。

二、Flink中的状态管理

按照数据的划分和扩张方式，Flink中大致分为2类：

Keyed States：记录每个Key对应的状态值一个Task上可能包含多个Key不同Task上不会出现相同的Key ，常用的 MapState, ValueState

Operator States：记录每个Task对应的状态值数据类型

ListState：并发度在改变的时候，会将并发上的每个List都取出，然后把这些List合并到一个新的List,然后根据元素的个数在均匀分配给新的Task;
UnionListState：相比于ListState更加灵活，把划分的方式交给用户去做，当改变并发的时候，会将原来的List拼接起来。然后不做划分，直接交给用户；
BroadcastState：如大表和小表做Join时，小表可以直接广播给大表的分区，在每个并发上的数据都是完全一致的。做的更新也相同，当改变并发的时候，把这些数据COPY到新的Task即可

state 存储在 State Backend 中，State Backend 一共有三种：

MemoryStateBackend
FsStateBackend
RocksDBStateBackend。但是最终都要持久化到磁盘上更保险，一般存储在hdfs中，那么 flink 的 Keyed States 和 Operator States 如何存储在 hdfs 中的呢？

三、State 在 hdfs 中的存储格式

Keyed States 和 Operator States 会存储再一个带有编号的 chk 目录中，比如说一个 flink 任务的 Keyed States 的 subTask 个数是4，Operator States 对应的 subTask 也是 4，那么 chk 会存一个元数据文件 _metadata ，四个 Keyed States 文件，四个 Operator States 的文件。

也就是说 Keyed States 和 Operator States 会分别存储 subTask 总数个状态文件。

四、State存在形式

Keyed State 和 Operator State 存在两种形式：managed （托管状态）和 raw（原始状态）。

托管状态是由Flink框架管理的状态；而原始状态是由用户自行管理状态的具体数据结构，框架在做checkpoint的时候，使用bytes 数组读写状态内容，对其内部数据结构一无所知。

通常所有的 datastream functions 都可以使用托管状态，但是原始状态接口仅仅能够在实现 operators 的时候使用。

推荐使用 managed state 而不是使用 raw state，因为使用托管状态的时候 Flink 可以在 parallelism 发生改变的情况下能够动态重新分配状态，而且还能更好的进行内存管理。

五、State的使用

其实 flink 中的 sum()，reduce() 等聚合算子都是有状态的计算，我们不妨看看他们的源码，里面肯定使用了 Keyed States 或者Operator States，我们以 sum() 算子源码举例，sum 底层调用了 StreamGroupedReduce，看到源码发现真的使用了ValueState，相信看完源码的你已经会自定义带有状态的算子了。

public class StreamGroupedReduce<IN> extends AbstractUdfStreamOperator<IN, ReduceFunction<IN>>
  implements OneInputStreamOperator<IN, IN> {
 private static final long serialVersionUID = 1L;
 private static final String STATE_NAME = "_op_state";
 private transient ValueState<IN> values;
 private TypeSerializer<IN> serializer;
 public StreamGroupedReduce(ReduceFunction<IN> reducer, TypeSerializer<IN> serializer) {
  super(reducer);
  this.serializer = serializer;
 }
 @Override
 public void open() throws Exception {
  super.open();
  ValueStateDescriptor<IN> stateId = new ValueStateDescriptor<>(STATE_NAME, serializer);
  values = getPartitionedState(stateId);
 }
 @Override
 public void processElement(StreamRecord<IN> element) throws Exception {
  IN value = element.getValue();
  IN currentValue = values.value();
        //如果任务不是第一次启动，也就是hdfs中已经存储过中间状态
  if (currentValue != null) {
   IN reduced = userFunction.reduce(currentValue, value);
   values.update(reduced);
   output.collect(element.replace(reduced));
  } 
        //如果任务第一次启动
        else {
   values.update(value);
   output.collect(element.replace(value));
  }
 }
}

六、State 过期时间TTL

使用 flink 进行实时计算中，会遇到一些状态数不断累积，导致状态量越来越大的情形。

例如，作业中定义了超长的时间窗口，或者在动态表上应用了无限范围的 GROUP BY 语句，以及执行了没有时间窗口限制的双流 JOIN 等等操作。

对于这些情况，经常导致堆内存出现 OOM，或者堆外内存（RocksDB）用量持续增长导致超出容器的配额上限，造成作业的频繁崩溃。从 Flink 1.6 版本开始引入了 State TTL 特性，该特性可以允许对作业中定义的 Keyed 状态进行超时自动清理，对于Table API 和 SQL 模块引入了空闲状态保留时间（Idle State Retention Time）进行状态管理，下面我们具体介绍一下。

1、State TTL 功能的用法

在 Flink 的官方文档中给我们展示了State TTL的基本用法，用法示例如下：

import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
 
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);

可以看到，要使用 State TTL 功能，首先要定义一个 StateTtlConfig 对象。这个 StateTtlConfig 对象可以通过构造器模式（Builder Pattern）来创建，典型地用法是传入一个 Time 对象作为 TTL 时间，然后设置更新类型（Update Type）和状态可见性（State Visibility），这两个功能的含义将在下面的文章中详细描述。当 StateTtlConfig 对象构造完成后，即可在后续声明的状态描述符（State Descriptor）中启用 State TTL 功能了。

从上述的代码也可以看到，State TTL 功能所指定的过期时间并不是全局生效的，而是和某个具体的状态所绑定。换而言之，如果希望对所有状态都生效，那么就需要对所有用到的状态定义都传入 StateTtlConfig 对象。对 Flink 源码感兴趣的同学，可以尝试为 Flink 增加一个默认的 StateTTL 选项，实现起来很简单，这里不再展开说明了。

State TTL 使用的更多案例，可以参见官方的 flink-stream-state-ttl-test 包，它提供了很多测试用例可以参考。

2、StateTtlConfig 的参数说明

TTL：表示状态的过期时间，是一个 org.apache.flink.api.common.time.Time 对象。一旦设置了 TTL，那么如果上次访问的时间戳 + TTL 超过了当前时间，则表明状态过期了（这是一个简化的说法，严谨的定义请参考 org.apache.flink.runtime.state.ttl.TtlUtils 类中关于 expired 的实现）。
UpdateType：表示状态时间戳的更新的时机，是一个 Enum 对象。如果设置为 Disabled，则表明不更新时间戳；如果设置为 OnCreateAndWrite，则表明当状态创建或每次写入时都会更新时间戳；如果设置为 OnReadAndWrite，则除了在状态创建和写入时更新时间戳外，读取也会更新状态的时间戳。
StateVisibility：表示对已过期但还未被清理掉的状态如何处理，也是 Enum 对象。如果设置为 ReturnExpiredIfNotCleanedUp，那么即使这个状态的时间戳表明它已经过期了，但是只要还未被真正清理掉，就会被返回给调用方；如果设置为 NeverReturnExpired，那么一旦这个状态过期了，那么永远不会被返回给调用方，只会返回空状态，避免了过期状态带来的干扰。
TimeCharacteristic 以及 TtlTimeCharacteristic：表示 State TTL 功能所适用的时间模式，仍然是 Enum 对象。前者已经被标记为 Deprecated（废弃），推荐新代码采用新的 TtlTimeCharacteristic 参数。截止到 Flink 1.8，只支持 ProcessingTime 一种时间模式，对 EventTime 模式的 State TTL 支持还在开发中。
CleanupStrategies：表示过期对象的清理策略，目前来说有三种 Enum 值。当设置为 FULL_STATE_SCAN_SNAPSHOT 时，对应的是 EmptyCleanupStrategy 类，表示对过期状态不做主动清理，当执行完整快照（Snapshot / Checkpoint）时，会生成一个较小的状态文件，但本地状态并不会减小。唯有当作业重启并从上一个快照点恢复后，本地状态才会实际减小，因此可能仍然不能解决内存压力的问题。为了应对这个问题，Flink 还提供了增量清理的枚举值，分别是针对 Heap StateBackend 的 INCREMENTAL_CLEANUP（对应 IncrementalCleanupStrategy 类），以及对 RocksDB StateBackend 有效的 ROCKSDB_COMPACTION_FILTER（对应 RocksdbCompactFilterCleanupStrategy 类）。对于增量清理功能，Flink 可以被配置为每读取若干条记录就执行一次清理操作，而且可以指定每次要清理多少条失效记录；对于 RocksDB 的状态清理，则是通过 JNI 来调用 C++ 语言编写的 FlinkCompactionFilter 来实现，底层是通过 RocksDB 提供的后台 Compaction 操作来实现对失效状态过滤的。

配置中有下面几个配置项可以选择：StateTtlConfig中的newBuilder这个方法是必须的，它是设置生存周期的值。

TTL 刷新策略（默认OnCreateAndWrite）

策略类型	描述
StateTtlConfig.UpdateType.Disabled	禁用TTL，永不过期
StateTtlConfig.UpdateType.OnCreateAndWrite	每次写操作都会更新State的最后访问时间
StateTtlConfig.UpdateType.OnReadAndWrite	每次读写操作都会跟新State的最后访问时间

状态可见性（默认NeverReturnExpired）

策略类型	描述
StateTtlConfig.StateVisibility.NeverReturnExpired	永不返回过期状态
StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp	可以返回过期但尚未被清理的状态值

3、Notes:

状态后端存储最后一次修改的时间戳和值，这意味着启用该特性会增加状态存储的消耗。堆状态后端在内存中存储一个附加的Java对象，其中包含对用户状态对象的引用和一个原始长值。RocksDB状态后端为每个存储值、列表条目或映射条目添加8个字节;
目前只支持与处理时间相关的TTLs;
如果试图使用启用TTL的描述符或使用启用TTL的描述符恢复先前在没有TTL的情况下配置的状态，将导致兼容性失败和statmigration异常;
TTL配置不是check- or savepoints的一部分，而是Flink在当前运行的作业中如何处理它的一种方式

七、State清除策略

1、Cleanup in full snapshot

默认情况下，过期值只有在显式读出时才会被删除，例如通过调用 ValueState.value() 方法。

此外，您可以在获取完整状态快照时激活清理操作，这将减少其大小。

在当前实现下，本地状态不会被清除，但在从前一个快照恢复时，它不会包含已删除的过期状态。可以在StateTtlConfig 中配置。（1）下面的配置选项不适用于 RocksDB state backend上的 increamental checkpointing；（2）对于现有作业，此清理策略可以在 StateTtlConfig 中随时激活或停用，例如从保存点重新启动后可以使用。

import org.apache.flink.api.common.state.StateTtlConfig
import org.apache.flink.api.common.time.Time
val ttlConfig = StateTtlConfig
    .newBuilder(Time.seconds(1))
    .cleanupFullSnapshot
    .build

2、Incremental cleanup

另一个选项是增量地触发对某些状态项的清理。触发器可以是来自每个状态访问或/和每个记录处理的回调。如果这个清理策略在某个状态下活跃的，那么存储后端会在其所有条目上为该状态保留一个惰性全局迭代器。

每次触发增量清理时，迭代器都会被提升。检查遍历的状态项，并清理过期的状态项。这个特性可以在StateTtlConfig中激活：

import org.apache.flink.api.common.state.StateTtlConfig
val ttlConfig = StateTtlCon fig
    .newBuilder(Time.seconds(1))
    .cleanupIncrementally
    .build

上面的策略有两个参数，第一个参数：第是每次清理触发的检查状态的条件。如果启用，则每次状态访问都将触发它。第二个参数：是否为每个记录处理额外触发清理。Notes：

如果对状态没有访问或者没有任何处理的记录，那么状态会一直保留；
增量状态的清理增加了记录处理的延迟；
目前，增量状态的清理策略仅仅在对堆状态后端被实现了，对于设置了RocksDB的将没有效果；
如果使用堆状态后端进行同步快照，全局迭代器在跌倒时会保留所有键的副本，因为它的特性不支持对并发数的修改。使用此功能将增加内存消耗。异步快照进行对状态的保存就没有这种情况发生；
对于现有的作业，可以通过在StateTtlConfig中设置这种清理策略能够随时被激活和停用，例如：从保存点重新启动后。

3、Cleanup during RocksDB compaction

如果使用RocksDB进行状态的管理，另一个清理策略就是激活Flink的压缩过滤这个策略。RocksDB会定期使用异步压缩来合并状态的更新和减少储存。Flink压缩过滤器使用TTL检查状态的过期时间戳，并排除过期值。

默认情况下是关闭该特性的。对于RocksDB进行状态管理首先要做的就是要激活，通过Flink配置文件配置state.backend.rocksdb.ttl.compaction.filter.enabled，或者对于一个Flink job来说如果一个自定义的RocksDB 状态管理被创建那么它可以调用 RocksDBStateBackend::enableTtlCompactionFilter。

然后任何带有TTL的状态都可以配置来去使用过滤器。

import org.apache.flink.api.common.state.StateTtlConfig
val ttlConfig = StateTtlConfig
    .newBuilder(Time.seconds(1))
    .cleanupInRocksdbCompactFilter
    .build

RocksDB compaction filter将会从Flink每次处理完一定数据量的状态之后，从Flink查询用于检查过期的当前时间戳,这个数字默认是1000。你也可以选择更改它，并将自定义值传递给StateTtlConfig.newBuilder(…)。cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries)方法。频繁的跟新时间错可以提高清理的数据但是会降低压缩性能，因为它使用了来自本地的JNI的调用。

Notes:

在压缩过程中调用TTL过滤器会减慢它的速度。TTL过滤器必须解析上次访问的时间戳，并检查每个正在压缩的键的每个存储状态条目的过期时间。对于集合状态类型(列表或映射)，每个存储的元素也调用该检查;
对于现有作业，此清理策略可以在StateTtlConfig中随时激活或停用，例如从保存点重新启动后。

目前，管理 operator state 仅仅支持使用 List 类型。当前，支持 List 样式的托管运算符状态,彼此之间相互独立，因此可以在重新缩放时可以重新分配。换句话说，这些对象是可以重新分配 non-keyed state 的最佳粒度。根据状态访问方法，定义一下重新分配方案。

八、Table API 和 SQL 模块状态管理

上面介绍了Flink State TTL 机制，这项机制对于应对通用的状态暴增特别有效。

然而，这个特性也有其缺陷，例如不能保证一定可以及时清理掉失效的状态，以及目前仅支持 Processing Time 时间模式等等，另外对于旧版本的 Flink（1.6 之前），State TTL 功能也无法使用。

针对 Table API 和 SQL 模块的持续查询/聚合语句，Flink 还提供了另一项失效状态清理机制，这就是本文要提到的 Idle State Retention Time 选项，Flink 很早就提供了这个选项，该特性是借助 Query Configuration 配置项来定义的，但很多人并未启用，也不理解其中隐藏的暗坑。本文将对这一特性做说明，并给出一些使用建议。

1、问题引入

同样以官网文档的案例为起点，一个持续查询的 GROUP BY 语句，它没有时间窗口的定义，理论上会无限地计算下去：

SELECT sessionId, COUNT(*) FROM clicks GROUP BY sessionId;

这就带来了一个问题：随着时间的不断推进，内存中积累的状态会越来越多，因为数据流是无穷无尽、持续流入的，Flink 并不知道如何丢弃旧的数据。在这种情况下，如果放任不管，那么迟早有一天作业的状态数达到了存储系统的容量极限，从而造成作业的崩溃。

针对这个问题，Flink 提出了空闲状态保留时间（Idle State Retention Time）的概念。

通过为每个状态设置 Timer，如果这个状态中途被访问过，则重新设置 Timer；否则（如果状态一直未被访问，长期处于 Idle 状态）则在 Timer 到期时做状态清理。

这样，就可以确保每个状态都能得到及时的清理。

通过调用 StreamQueryConfig 的 withIdleStateRetentionTime 方法，可以为这个 QueryConfig 对象设置最小和最大的清理周期。

这样，Flink 可以保证最早和最晚的状态清理时间。

需要注意的是，旧版本 Flink 允许只指定一个参数，表示最早和最晚清理周期相同，但是这样可能会导致同一时间段有很多状态都到期，从而造成瞬间的处理压力。

新版本的 Flink 要求两个参数之间的差距至少要达到 5 分钟，从而避免大量状态瞬间到期，对系统造成的冲击。

StreamTableEnvironment.getConfig().setIdleStateRetentionTime(min, max);
或者
StreamQueryConfig qConfig = ...
// set idle state retention time: min = 12 hours, max = 24 hours
qConfig.withIdleStateRetentionTime(Time.hours(12), Time.hours(24));

注意：默认情况下 StreamQueryConfig 的设置并不是全局的。因此当设置了清理周期以后，需要在 StreamTableEnvironment 类调用 toAppendStream 或 toRetractStream 将 Table 转为 DataStream 时，显式传入这个 QueryConfig 对象作为参数，才可以令该功能生效。

为什么会有这种限制呢？看一下源码就知道了。

2、如何实现的

idle state retention time特性在底层以o.a.f.table.runtime.functions.CleanupState接口来表示，代码如下。

public interface CleanupState {
    default void registerProcessingCleanupTimer(
            ValueState<Long> cleanupTimeState,
            long currentTime,
            long minRetentionTime,
            long maxRetentionTime,
            TimerService timerService)
            throws Exception {
        // last registered timer
        Long curCleanupTime = cleanupTimeState.value();
 
        // check if a cleanup timer is registered and
        // that the current cleanup timer won't delete state we need to keep
        if (curCleanupTime == null || (currentTime + minRetentionTime) > curCleanupTime) {
            // we need to register a new (later) timer
            long cleanupTime = currentTime + maxRetentionTime;
            // register timer and remember clean-up time
            timerService.registerProcessingTimeTimer(cleanupTime);
            // delete expired timer
            if (curCleanupTime != null) {
                timerService.deleteProcessingTimeTimer(curCleanupTime);
            }
            cleanupTimeState.update(cleanupTime);
        }
    }
}

由上可知，每个key对应的最近状态清理时间会单独维护在ValueState中。如果满足以下两条件之一：

ValueState为空（即这个key是第一次出现）
或者当前时间加上minRetentionTime已经超过了最近清理的时间

就用当前时间加上maxRetentionTime注册新的Timer，并将其时间戳存入ValueState，用于触发下一次清理。如果有已经过期了的Timer，则一并删除之。可见，如果minRetentionTime和maxRetentionTime的间隔设置太小，就会比较频繁地产生Timer与更新ValueState，维护Timer的成本会变大

新版本的 Flink 提供了一个 QueryConfigProvider 类（它实现了 PlannerConfig 接口，允许嵌入一个 StreamQueryConfig 对象），可以通过对 TableConfig 设置 PlannerConfig 的方式（调用 addPlannerConfig 方法），来传入设置好 StreamQueryConfig 对象的 QueryConfigProvider. 这样，当 StreamPlanner 将定义的 Table 翻译为 Plan 时，可以自动使用之前定义的 StreamQueryConfig，从而实现全局的 StreamQueryConfig 设定。对于旧的 Flink 版本，只能通过修改源码的方式来设置，较为繁琐。