Flink（十）：DataStream API (七) 状态

1. 状态的定义

在 Apache Flink 中，状态（State） 是指在数据流处理过程中需要持久化和追踪的中间数据，它允许 Flink 在处理事件时保持上下文信息，从而支持复杂的流式计算任务，如聚合、窗口计算、联接等。状态是 Flink 处理有状态操作（如窗口、时间戳操作、聚合等）的核心组成部分。

2. 状态的类型

Flink 提供了强大的状态管理机制，允许应用程序在分布式环境中处理状态，保证高可用性和容错性。Flink 的状态分为 Keyed State 和 Operator State，并提供了不同的存储和恢复机制。

2.1 Keyed State（按键状态）

Keyed State 是基于流中每个元素的键进行管理的状态。每个键会有一个独立的状态，这对于需要按照每个输入元素的唯一标识符（如用户 ID、商品 ID 等）维护状态的操作非常有用。
Keyed State 主要用于需要对流中的每个“键”进行独立计算的场景，如按用户进行会话计算、按时间窗口聚合等。

常见的 Keyed State 类型包括：

ValueState：存储与每个键相关联的单个值。
ListState：存储与每个键相关联的多个值，通常用来表示一个元素列表。
MapState：存储与每个键相关联的键值对，适用于需要维护多个关联数据的场景。
ReducingState：支持对每个键的值进行累加或其他聚合操作。
AggregatingState：可以根据给定的聚合函数对每个键的状态进行聚合。

2.2 Operator State（操作符状态）

Operator State 是由操作符（如 Flink 中的算子）管理的状态，通常用于保持操作符内部的状态信息，不与键相关联。它用于管理一些需要跨整个流处理作业的全局状态，如窗口管理、算子内部缓冲区等。
Operator State 主要用于在分布式环境中处理算子级别的状态，尤其在对状态进行恢复时非常重要，帮助 Flink 恢复作业。

常见的 Operator State 类型包括：

ListState：与键无关，存储多个值。
UnionListState : 与键无关，存储多个值, UnionListState 是 ListState 的扩展，主要用于 跨多个并行实例 共享状态。在 Flink 的 流式应用程序 中，如果多个并行实例需要访问和修改共享的状态，通常使用 UnionListState。
BroadcastState：存储和广播信息。

3. Keyed State

Keyed 状态可以看作是一个嵌入式的键值存储。该状态是与由有状态操作符读取的流一起严格地进行分区和分布的。因此，只有在 Keyed 流 上才能访问键值状态，也就是说，只有在进行键控/分区数据交换后，才能访问与当前事件的键相关联的值。将流的键与状态对齐确保了所有的状态更新都是本地操作，从而在没有事务开销的情况下保证一致性。这个对齐还使得 Flink 能够透明地重新分配状态并调整流的分区。

Keyed 状态进一步组织为所谓的 Key Groups（键组）。Key Groups 是 Flink 重新分配 Keyed 状态的最小单位；其数量与定义的最大并行度相同。在执行过程中，每个并行实例的键控操作符都处理一个或多个 Key Groups 中的键。

3.1 使用 Keyed State

keyed state 接口提供不同类型状态的访问接口，这些状态都作用于当前输入数据的 key 下。换句话说，这些状态仅可在 KeyedStream 上使用，在Java/Scala API上可以通过 stream.keyBy(...) 得到 KeyedStream，在Python API上可以通过 stream.key_by(...) 得到 KeyedStream。

接下来，我们会介绍不同类型的状态，然后介绍如何使用他们。所有支持的状态类型如下所示：

ValueState<T>: 保存一个可以更新和检索的值（如上所述，每个值都对应到当前的输入数据的 key，因此算子接收到的每个 key 都可能对应一个值）。这个值可以通过 update(T) 进行更新，通过 T value() 进行检索。
ListState<T>: 保存一个元素的列表。可以往这个列表中追加数据，并在当前的列表上进行检索。可以通过 add(T) 或者 addAll(List<T>) 进行添加元素，通过 Iterable<T> get() 获得整个列表。还可以通过 update(List<T>) 覆盖当前的列表。
ReducingState<T>: 保存一个单值，表示添加到状态的所有值的聚合。接口与 ListState 类似，但使用 add(T) 增加元素，会使用提供的 ReduceFunction 进行聚合。
AggregatingState<IN, OUT>: 保留一个单值，表示添加到状态的所有值的聚合。和 ReducingState 相反的是, 聚合类型可能与添加到状态的元素的类型不同。接口与 ListState 类似，但使用 add(IN) 添加的元素会用指定的 AggregateFunction 进行聚合。
MapState<UK, UV>: 维护了一个映射列表。你可以添加键值对到状态中，也可以获得反映当前所有映射的迭代器。使用 put(UK，UV) 或者 putAll(Map<UK，UV>) 添加映射。使用 get(UK) 检索特定 key。使用 entries()，keys() 和 values() 分别检索映射、键和值的可迭代视图。你还可以通过 isEmpty() 来判断是否包含任何键值对。

所有类型的状态还有一个clear() 方法，清除当前 key 下的状态数据，也就是当前输入元素的 key。

请牢记，这些状态对象仅用于与状态交互。状态本身不一定存储在内存中，还可能在磁盘或其他位置。另外需要牢记的是从状态中获取的值取决于输入元素所代表的 key。因此，在不同 key 上调用同一个接口，可能得到不同的值。

你必须创建一个 StateDescriptor，才能得到对应的状态句柄。这保存了状态名称（正如我们稍后将看到的，你可以创建多个状态，并且它们必须具有唯一的名称以便可以引用它们），状态所持有值的类型，并且可能包含用户指定的函数，例如ReduceFunction。根据不同的状态类型，可以创建ValueStateDescriptor，ListStateDescriptor， AggregatingStateDescriptor, ReducingStateDescriptor 或 MapStateDescriptor。

状态通过 RuntimeContext 进行访问，因此只能在 rich functions 中使用。RichFunction 中 RuntimeContext 提供如下方法：

ValueState<T> getState(ValueStateDescriptor<T>)
ReducingState<T> getReducingState(ReducingStateDescriptor<T>)
ListState<T> getListState(ListStateDescriptor<T>)
AggregatingState<IN, OUT> getAggregatingState(AggregatingStateDescriptor<IN, ACC, OUT>)
MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV>)

下面是一个 FlatMapFunction 的例子，展示了如何将这些部分组合起来：

public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {

    /**
     * The ValueState handle. The first field is the count, the second field a running sum.
     */
    private transient ValueState<Tuple2<Long, Long>> sum;

    @Override
    public void flatMap(Tuple2<Long, Long> input, Collector<Tuple2<Long, Long>> out) throws Exception {

        // access the state value
        Tuple2<Long, Long> currentSum = sum.value();

        // update the count
        currentSum.f0 += 1;

        // add the second field of the input value
        currentSum.f1 += input.f1;

        // update the state
        sum.update(currentSum);

        // if the count reaches 2, emit the average and clear the state
        if (currentSum.f0 >= 2) {
            out.collect(new Tuple2<>(input.f0, currentSum.f1 / currentSum.f0));
            sum.clear();
        }
    }

    @Override
    public void open(OpenContext ctx) {
        ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
                new ValueStateDescriptor<>(
                        "average", // the state name
                        TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {}), // type information
                        Tuple2.of(0L, 0L)); // default value of the state, if nothing was set
        sum = getRuntimeContext().getState(descriptor);
    }
}

// this can be used in a streaming program like this (assuming we have a StreamExecutionEnvironment env)
env.fromElements(Tuple2.of(1L, 3L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L), Tuple2.of(1L, 4L), Tuple2.of(1L, 2L))
        .keyBy(value -> value.f0)
        .flatMap(new CountWindowAverage())
        .print();

// the printed output will be (1,4) and (1,5)

3.2 状态有效期 (TTL)

任何类型的 keyed state 都可以有 有效期 (TTL)。如果配置了 TTL 且状态值已过期，则会尽最大可能清除对应的值，所有状态类型都支持单元素的 TTL。这意味着列表元素和映射元素将独立到期。在使用状态 TTL 前，需要先构建一个StateTtlConfig 配置对象。然后把配置传递到 state descriptor 中启用 TTL 功能：

import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import java.time.Duration;

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

TTL 配置有以下几个选项： newBuilder 的第一个参数表示数据的有效期，是必选项。TTL 的更新策略（默认是 OnCreateAndWrite）:

StateTtlConfig.UpdateType.OnCreateAndWrite - 仅在创建和写入时更
StateTtlConfig.UpdateType.OnReadAndWrite - 读取时也更新

数据在过期但还未被清理时的可见性配置如下（默认为 NeverReturnExpired):

StateTtlConfig.StateVisibility.NeverReturnExpired - 不返回过期数据
StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp - 会返回过期但未清理的数据

NeverReturnExpired 情况下，过期数据就像不存在一样，不管是否被物理删除。这对于不能访问过期数据的场景下非常有用，比如敏感数据。 ReturnExpiredIfNotCleanedUp 在数据被物理删除前都会返回。

注意:

状态上次的修改时间会和数据一起保存在 state backend 中，因此开启该特性会增加状态数据的存储。 Heap state backend 会额外存储一个包括用户状态以及时间戳的 Java 对象，RocksDB state backend 会在每个状态值（list 或者 map 的每个元素）序列化后增加 8 个字节。
暂时只支持基于 processing time 的 TTL。
尝试从 checkpoint/savepoint 进行恢复时，TTL 的状态（是否开启）必须和之前保持一致，否则会遇到 “StateMigrationException”。
TTL 的配置并不会保存在 checkpoint/savepoint 中，仅对当前 Job 有效。
不建议checkpoint恢复前后将state TTL从短调长，这可能会产生潜在的数据错误。
当前开启 TTL 的 map state 仅在用户值序列化器支持 null 的情况下，才支持用户值为 null。如果用户值序列化器不支持 null，可以用 NullableSerializer 包装一层。
启用 TTL 配置后，StateDescriptor 中的 defaultValue（已被标记 deprecated）将会失效。这个设计的目的是为了确保语义更加清晰，在此基础上，用户需要手动管理那些实际值为 null 或已过期的状态默认值。

3.2.1 过期数据的清理

默认情况下，过期数据会在读取的时候被删除，例如 ValueState#value，同时会有后台线程定期清理（如果 StateBackend 支持的话）。可以通过 StateTtlConfig 配置关闭后台清理：

import org.apache.flink.api.common.state.StateTtlConfig;

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Duration.ofSeconds(1))
    .disableCleanupInBackground()
    .build();

可以按照如下所示配置更细粒度的后台清理策略。当前的实现中 HeapStateBackend 依赖增量数据清理，RocksDBStateBackend 利用压缩过滤器进行后台清理。

3.2.2 全量快照时进行清理

另外，你可以启用全量快照时进行清理的策略，这可以减少整个快照的大小。当前实现中不会清理本地的状态，但从上次快照恢复时，不会恢复那些已经删除的过期数据。该策略可以通过 StateTtlConfig 配置进行配置：

import org.apache.flink.api.common.state.StateTtlConfig;
import java.time.Duration;

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Duration.ofSeconds(1))
    .cleanupFullSnapshot()
    .build();

这种策略在 RocksDBStateBackend 的增量 checkpoint 模式下无效。

注意:这种清理方式可以在任何时候通过 StateTtlConfig 启用或者关闭，比如在从 savepoint 恢复时。

3.2.3 增量数据清理

另外可以选择增量式清理状态数据，在状态访问或/和处理时进行。如果某个状态开启了该清理策略，则会在存储后端保留一个所有状态的惰性全局迭代器。每次触发增量清理时，从迭代器中选择已经过期的数进行清理。该特性可以通过 StateTtlConfig 进行配置：

import org.apache.flink.api.common.state.StateTtlConfig;
 StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Duration.ofSeconds(1))
    .cleanupIncrementally(10, true)
    .build();

该策略有两个参数。第一个是每次清理时检查状态的条目数，在每个状态访问时触发。第二个参数表示是否在处理每条记录时触发清理。 Heap backend 默认会检查 5 条状态，并且关闭在每条记录时触发清理。

注意:

如果没有 state 访问，也没有处理数据，则不会清理过期数据。
增量清理会增加数据处理的耗时。
现在仅 Heap state backend 支持增量清除机制。在 RocksDB state backend 上启用该特性无效。
如果 Heap state backend 使用同步快照方式，则会保存一份所有 key 的拷贝，从而防止并发修改问题，因此会增加内存的使用。但异步快照则没有这个问题。
对已有的作业，这个清理方式可以在任何时候通过 StateTtlConfig 启用或禁用该特性，比如从 savepoint 重启后.

3.2.4 在 RocksDB 压缩时清理

如果使用 RocksDB state backend，则会启用 Flink 为 RocksDB 定制的压缩过滤器。RocksDB 会周期性的对数据进行合并压缩从而减少存储空间。 Flink 提供的 RocksDB 压缩过滤器会在压缩时过滤掉已经过期的状态数据。该特性可以通过 StateTtlConfig 进行配置：

import org.apache.flink.api.common.state.StateTtlConfig;

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Duration.ofSeconds(1))
    .cleanupInRocksdbCompactFilter(1000, Duration.ofHours(1))
    .build();

Flink 处理一定条数的状态数据后，会使用当前时间戳来检测 RocksDB 中的状态是否已经过期，你可以通过

StateTtlConfig.newBuilder(...).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries) 方法指定处理状态的条数。时间戳更新的越频繁，状态的清理越及时，但由于压缩会有调用 JNI 的开销，因此会影响整体的压缩性能。 RocksDB backend 的默认后台清理策略会每处理 1000 条数据进行一次。定期压缩可以加速过期状态条目的清理，特别是对于很少访问的状态条目。比这个值早的文件将被选取进行压缩，并重新写入与之前相同的 Level 中。该功能可以确保文件定期通过压缩过滤器压缩。您可以通过StateTtlConfig.newBuilder(...).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries, Duration periodicCompactionTime) 方法设定定期压缩的时间。定期压缩的时间的默认值是 30 天。您可以将其设置为 0 以关闭定期压缩或设置一个较小的值以加速过期状态条目的清理，但它将会触发更多压缩。还可以通过配置开启 RocksDB 过滤器的 debug 日志： log4j.logger.org.rocksdb.FlinkCompactionFilter=DEBUG

注意:

压缩时调用 TTL 过滤器会降低速度。TTL 过滤器需要解析上次访问的时间戳，并对每个将参与压缩的状态进行是否过期检查。对于集合型状态类型（比如 list 和 map），会对集合中每个元素进行检查。
对于元素序列化后长度不固定的列表状态，TTL 过滤器需要在每次 JNI 调用过程中，额外调用 Flink 的 java 序列化器，从而确定下一个未过期数据的位置。
对已有的作业，这个清理方式可以在任何时候通过 StateTtlConfig 启用或禁用该特性，比如从 savepoint 重启后。
定期压缩功能只在 TTL 启用时生效。

4. Operator State

算子状态（或者非 keyed 状态）是绑定到一个并行算子实例的状态。Kafka Connector 是 Flink 中使用算子状态一个很具有启发性的例子。Kafka consumer 每个并行实例维护了 topic partitions 和偏移量的 map 作为它的算子状态。当并行度改变的时候，算子状态支持将状态重新分发给各并行算子实例。处理重分发过程有多种不同的方案。在典型的有状态 Flink 应用中你无需使用算子状态。它大都作为一种特殊类型的状态使用。用于实现 source/sink，以及无法对 state 进行分区而没有主键的这类场景中。

4.1 广播状态 (Broadcast State)

广播状态是一种特殊的算子状态。引入它的目的在于支持一个流中的元素需要广播到所有下游任务的使用情形。在这些任务中广播状态用于保持所有子任务状态相同。该状态接下来可在第二个处理记录的数据流中访问。可以设想包含了一系列用于处理其他流中元素规则的低吞吐量数据流，这个例子自然而然地运用了广播状态。考虑到上述这类使用情形，广播状态和其他算子状态的不同之处在于：

它具有 map 格式，
它仅在一些特殊的算子中可用。这些算子的输入为一个广播数据流和非广播数据流，
这类算子可以拥有不同命名的多个广播状态 。

4.2 使用 Operator State

用户可以通过实现 CheckpointedFunction 接口来使用 operator stateCheckpointedFunction 接口提供了访问 non-keyed state 的方法，需要实现如下两个方法：

void snapshotState(FunctionSnapshotContext context) throws Exception;

void initializeState(FunctionInitializationContext context) throws Exception;

进行 checkpoint 时会调用 snapshotState()。用户自定义函数初始化时会调用 initializeState()，初始化包括第一次自定义函数初始化和从之前的 checkpoint 恢复。因此 initializeState() 不仅是定义不同状态类型初始化的地方，也需要包括状态恢复的逻辑。当前 operator state 以 list 的形式存在。这些状态是一个 可序列化 对象的集合 List，彼此独立，方便在改变并发后进行状态的重新分派。换句话说，这些对象是重新分配 non-keyed state 的最细粒度。根据状态的不同访问方式，有如下几种重新分配的模式：

Even-split redistribution: 每个算子都保存一个列表形式的状态集合，整个状态由所有的列表拼接而成。当作业恢复或重新分配的时候，整个状态会按照算子的并发度进行均匀分配。比如说，算子 A 的并发读为 1，包含两个元素 element1 和 element2，当并发读增加为 2 时，element1 会被分到并发 0 上，element2 则会被分到并发 1 上。
Union redistribution: 每个算子保存一个列表形式的状态集合。整个状态由所有的列表拼接而成。当作业恢复或重新分配时，每个算子都将获得所有的状态数据。如果你的列表可能具有高基数，请不要使用此功能。检查点元数据将存储指向每个列表项的偏移量，这可能导致 RPC 帧大小或内存溢出错误。

下面的例子中的 SinkFunction 在 CheckpointedFunction 中进行数据缓存，然后统一发送到下游，这个例子演示了列表状态数据的 event-split redistribution。

public class BufferingSink
        implements SinkFunction<Tuple2<String, Integer>>,
                   CheckpointedFunction {

    private final int threshold;

    private transient ListState<Tuple2<String, Integer>> checkpointedState;

    private List<Tuple2<String, Integer>> bufferedElements;

    public BufferingSink(int threshold) {
        this.threshold = threshold;
        this.bufferedElements = new ArrayList<>();
    }

    @Override
    public void invoke(Tuple2<String, Integer> value, Context contex) throws Exception {
        bufferedElements.add(value);
        if (bufferedElements.size() >= threshold) {
            for (Tuple2<String, Integer> element: bufferedElements) {
                // send it to the sink
            }
            bufferedElements.clear();
        }
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        checkpointedState.update(bufferedElements);
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        ListStateDescriptor<Tuple2<String, Integer>> descriptor =
            new ListStateDescriptor<>(
                "buffered-elements",
                TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {}));

        checkpointedState = context.getOperatorStateStore().getListState(descriptor);

        if (context.isRestored()) {
            for (Tuple2<String, Integer> element : checkpointedState.get()) {
                bufferedElements.add(element);
            }
        }
    }
}

initializeState 方法接收一个 FunctionInitializationContext 参数，会用来初始化 non-keyed state 的 “容器”。这些容器是一个 ListState 用于在 checkpoint 时保存 non-keyed state 对象。注意这些状态是如何初始化的，和 keyed state 类似，StateDescriptor 会包括状态名字、以及状态类型相关信息。

ListStateDescriptor<Tuple2<String, Integer>> descriptor =
    new ListStateDescriptor<>(
        "buffered-elements",
        TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {}));

checkpointedState = context.getOperatorStateStore().getListState(descriptor);

调用不同的获取状态对象的接口，会使用不同的状态分配算法。比如

getUnionListState(descriptor) 会使用 union redistribution 算法，而 getListState(descriptor) 则简单的使用 even-split redistribution 算法。

当初始化好状态对象后，我们通过 isRestored() 方法判断是否从之前的故障中恢复回来，如果该方法返回 true 则表示从故障中进行恢复，会执行接下来的恢复逻辑。

正如代码所示，BufferingSink 中初始化时，恢复回来的 ListState 的所有元素会添加到一个局部变量中，供下次 snapshotState() 时使用。然后清空 ListState，再把当前局部变量中的所有元素写入到 checkpoint 中。另外，我们同样可以在 initializeState() 方法中使用 FunctionInitializationContext 初始化 keyed state。

4.3 带状态的 Source Function

带状态的数据源比其他的算子需要注意更多东西。为了保证更新状态以及输出的原子性（用于支持 exactly-once 语义），用户需要在发送数据前获取数据源的全局锁。

public static class CounterSource
        extends RichParallelSourceFunction<Long>
        implements CheckpointedFunction {

    /**  current offset for exactly once semantics */
    private Long offset = 0L;

    /** flag for job cancellation */
    private volatile boolean isRunning = true;
    
    /** 存储 state 的变量. */
    private ListState<Long> state;
     
    @Override
    public void run(SourceContext<Long> ctx) {
        final Object lock = ctx.getCheckpointLock();

        while (isRunning) {
            // output and state update are atomic
            synchronized (lock) {
                ctx.collect(offset);
                offset += 1;
            }
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        state = context.getOperatorStateStore().getListState(new ListStateDescriptor<>(
            "state",
            LongSerializer.INSTANCE));
            
        // 从我们已保存的状态中恢复 offset 到内存中，在进行任务恢复的时候也会调用此初始化状态的方法
        for (Long l : state.get()) {
            offset = l;
        }
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        state.update(Collections.singletonList(offset));
    }
}

希望订阅 checkpoint 成功消息的算子，可以参考

org.apache.flink.api.common.state.CheckpointListener 接口。

5. State Backends

键/值索引存储的具体数据结构取决于所选择的状态后端。一种状态后端将数据存储在内存中的哈希映射中，另一种状态后端则使用 RocksDB 作为键/值存储。除了定义存储状态的数据结构外，状态后端还实现了在某个时间点对键/值状态进行快照并将该快照作为检查点的一部分存储的逻辑。状态后端可以在不更改应用程序逻辑的情况下进行配置。

Flink 提供了多种 state backends，它用于指定状态的存储方式和位置。状态可以位于 Java 的堆或堆外内存。取决于你的 state backend，Flink 也可以自己管理应用程序的状态。为了让应用程序可以维护非常大的状态，Flink 可以自己管理内存（如果有必要可以溢写到磁盘）。默认情况下，所有 Flink Job 会使用 Flink 配置文件中指定的 state backend。但是，配置文件中指定的默认 state backend 会被 Job 中指定的 state backend 覆盖，如下所示。

Configuration config = new Configuration();
config.set(StateBackendOptions.STATE_BACKEND, "rocksdb");
env.configure(config);