Flink 状态管理

1.概述

算子任务不仅需要当前流的数据，还需要其他数据来参与计算，这里的其他数据就是状态。

一个算子任务会按照并行度分为多个并行子任务执行，而不同的子任务会占据不同的任务槽（task slot）。由于不同的 slot 在计算资源上是物理隔离的，所以Flink能管理的状态在并行任务间是无法共享的，每个状态只能针对当前子任务的实例有效。按键分组之后，任务所进行的所有计算都应该只针对当前 key 有效，所以状态也应该按照 key 彼此隔离。

无论是Keyed State还是Operator State，它们都是在本地实例上维护的，也就是说每个并行子任务维护着对应的状态，算子的子任务之间状态不共享。

2. 按键分组状态（Keyed State）

状态是根据输入流中定义的键（key）来维护和访问的状态，是以 key 为作用范围进行隔离的。

2.1 值状态

状态中只保存一个值。

• T value()：获取当前状态的值；
• update(T value)：对状态进行更新，传入的参数value就是要覆写的状态值。

ValueState<Integer> lastVcState;

lastVcState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("lastVcState", Types.INT));

int lastVc = lastVcState.value() == null ? 0 : lastVcState.value();

lastVcState.update(vc);

2.2 列表状态

将需要保存的数据，以列表（List）的形式组织起来。

• Iterable<T> get()：获取当前的列表状态，返回的是一个可迭代类型Iterable<T>；
• update(List<T> values)：传入一个列表values，直接对状态进行覆盖；
• add(T value)：在状态列表中添加一个元素value；
• addAll(List<T> values)：向列表中添加多个元素，以列表values形式传入。
• clear()：清空列表状态

2.3 Map 状态

把一些键值对（key-value）作为状态整体保存起来，可以认为就是一组key-value映射的列表。

• UV get(UK key)：传入一个key作为参数，查询对应的value值；
• put(UK key, UV value)：传入一个键值对，更新key对应的value值；
• putAll(Map<UK, UV> map)：将传入的映射map中所有的键值对，全部添加到映射状态中；
• remove(UK key)：将指定key对应的键值对删除；
• boolean contains(UK key)：判断是否存在指定的key，返回一个boolean值。
• Iterable<Map.Entry<UK, UV>> entries()：获取映射状态中所有的键值对；
• Iterable<UK> keys()：获取映射状态中所有的键（key），返回一个可迭代Iterable类型；
• Iterable<UV> values()：获取映射状态中所有的值（value），返回一个可迭代Iterable类型；
• boolean isEmpty()：判断映射是否为空，返回一个boolean值。

2.4 规约状态

类似于值状态（Value），不过需要对添加进来的所有数据进行归约，将归约聚合之后的值作为状态保存下来。ReducingState<T>这个接口调用的方法类似于ListState，只不过它保存的只是一个聚合值，所以调用.add()方法时，不是在状态列表里添加元素，而是直接把新数据和之前的状态进行归约，并用得到的结果更新状态。归约逻辑的定义，是在归约状态描述器（ReducingStateDescriptor）中，通过传入一个归约函数（ReduceFunction）来实现的。

2.5 聚合状态

与归约状态非常类似，聚合状态也是一个值，用来保存添加进来的所有数据的聚合结果。与ReducingState不同的是，它的聚合逻辑是由在描述器中传入一个更加一般化的聚合函数（AggregateFunction）来定义的；这也就是之前我们讲过的AggregateFunction，里面通过一个累加器（Accumulator）来表示状态，所以聚合的状态类型可以跟添加进来的数据类型完全不同，使用更加灵活。同样地，AggregatingState接口调用方法也与ReducingState相同，调用.add()方法添加元素时，会直接使用指定的AggregateFunction进行聚合并更新状态。

2.6 状态生存时间（TTL）

在实际应用中，很多状态会随着时间的推移逐渐增长，如果不加以限制，最终就会导致存储空间的耗尽。一个优化的思路是直接在代码中调用 clear() 方法去清除状态。另一个方法就是配置一个状态的“生存时间”（time-to-live，TTL），当状态在内存中存在的时间超出这个值时，就将它清除。状态创建的时候，设置 失效时间 = 当前时间 + TTL；之后如果有对状态的访问和修改，可以再对失效时间进行更新；当设置的清除条件被触发时（比如，状态被访问的时候，或者每隔一段时间扫描一次失效状态），就可以判断状态是否失效、从而进行清除了。

StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.seconds(10))

                                          .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)

                                          .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired).build();

 ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("my state", String.class);

 stateDescriptor.enableTimeToLive(ttlConfig);

newBuilder() 是状态TTL配置的构造器方法，必须调用，返回一个 Builder 之后再调用 build() 方法就可以得到 StateTtlConfig 了。方法需要传入一个 Time 作为参数，这就是设定的状态生存时间。

setUpdateType() 是设置更新类型。更新类型指定了什么时候更新状态失效时间，这里的 OnCreateAndWrite 表示只有创建状态和更改状态（写操作）时更新失效时间。另一种类型 OnReadAndWrite 则表示无论读写操作都会更新失效时间，也就是只要对状态进行了访问，就表明它是活跃的，从而延长生存时间。这个配置默认为 OnCreateAndWrite。

setStateVisibility() 设置状态的可见性。所谓的“状态可见性”，是指因为清除操作并不是实时的，所以当状态过期之后还有可能继续存在，这时如果对它进行访问，能否正常读取到就是一个问题了。这里设置的 NeverReturnExpired 是默认行为，表示从不返回过期值，也就是只要过期就认为它已经被清除了，应用不能继续读取；这在处理会话或者隐私数据时比较重要。对应的另一种配置是 ReturnExpireDefNotCleanedUp，就是如果过期状态还存在，就返回它的值。

3. 算子状态（Operator State）

算子状态是一个算子并行实例上定义的状态，作用范围被限定为当前算子任务。

Flink 支持：

• 重启时改并行度

• 从 Savepoint 恢复时改并行度

• 自动扩缩容（Adaptive Scheduler）

3.1 列表状态

与Keyed State中的ListState一样，将状态表示为一组数据的列表。

当算子并行度进行缩放调整时，算子的列表状态中的所有元素项会被统一收集起来，相当于把多个分区的列表合并成了一个“大列表”，然后再均匀地分配给所有并行任务。

1. 实现 CheckpointedFunction 接口
2. 本地变量持久化，重写 snapshotState 方法：将本地变量拷贝到算子状态中，开启 checkpoint 时才会调用
3. 初始化本地变量，重写 initializeState 方法：程序启动和恢复时，从状态中把数据添加到本地变量，每个子任务调用一次。

state = context
        .getOperatorStateStore()
        .getListState(new ListStateDescriptor<Long>("state", Types.LONG));

3.2 联合列表状态

与 ListState 类似，联合列表状态也会将状态表示为一个列表。它与常规列表状态的区别在于，算子并行度进行缩放调整时对于状态的分配方式不同。

UnionListState 的重点就在于“联合”（union）。在并行度调整时，常规列表状态是轮询分配状态项，而联合列表状态的算子则会直接广播状态的完整列表。这样，并行度缩放之后的并行子任务就获取到了联合后完整的“大列表”，可以自行选择要使用的状态项和要丢弃的状态项。

state = context
        .getOperatorStateStore()
        .getUnionListState(new ListStateDescriptor<Long>("state", Types.LONG));

3.3 广播状态

所有分区的所有数据都会访问到同一个状态，状态就像被“广播”到所有分区一样，这种算子状态，就叫作广播状态（BroadcastState）。

1）广播流的定义和配置

DataStreamSource<String> configDS = env.socketTextStream("hadoop102", 8888);

MapStateDescriptor<String, Integer> broadcastMapState = new MapStateDescriptor<>("broadcast-state", Types.STRING, Types.INT);

BroadcastStream<String> configBS = configDS.broadcast(broadcastMapState);

BroadcastConnectedStream<WaterSensor, String> sensorBCS = sensorDS.connect(configBS);

2）调用 Process 方法，参数是 BroadcastProcessFunction 的对象。重写方法 processElement (对广播状态可读权限)和 processBroadcastElement （对广播状态有修改权限）方法。

3.4 和 Keyed State 的核心区别

4. 状态后端

4.1 状态后端的分类

Flink中提供了两类不同的状态后端，一种是“哈希表状态后端”（HashMapStateBackend，默认），另一种是“内嵌RocksDB状态后端”（EmbeddedRocksDBStateBackend）。

4.2 状态后端的配置

4.2.1 配置默认的状态后端

在 flink-conf.yaml 中，可以使用 state.backend 来配置默认状态后端。

配置项为 hashmap，配置的就是 HashMapStateBackend；如果配置项的值是 rocksdb，这样配置的就是 EmbeddedRocksDBStateBackend。

# 默认状态后端

state.backend: hashmap

# 存放检查点的文件路径，定义了检查点和元数据写入的目录

state.checkpoints.dir: hdfs://xxxx:8020/flink/checkpoints

4.2.2 为每个作业（Per-job/Application）单独配置状态后端

env.setStateBackend(new HashMapStateBackend());

env.setStateBackend(new EmbeddedRocksDBStateBackend());