3.2、Flink流处理（Stream API）- State & Fault Tolerance（状态和容错）之 State的工作原理

今天药忘吃喽~ 2024-04-17 05:38 25阅读 0赞

**目录**

Keyed State and Operator State（两种基本State）

Raw and Managed State（两个基本状态的存在形式）

Using Managed Keyed State（如何使用）

状态有效期（TTL）

使用Managed Operator State

Stateful Source Functions（带状态的Source Functions）

--------------------

## Keyed State and Operator State（两种基本State） ##

Flink的两种基本State：Keyed State和Operator State。

**Keyed State**

Keyed State总是相对于键的，并且只能在KeyedStream上的函数和算子中使用。

您可以将Keyed State视为已分区或分片的算子状态，每个键只有一个状态分区。每个keyed-state在逻辑上都绑定到<parallel-operator-instance, key>的唯一组合，由于每个键“属于”keyed operator的一个并行实例，所以我们可以简单地将其看作<operator, key>。

Keyed State被进一步组织成所谓的Key Groups。Key Groups是一个原子单元，通过它，Flink可以重新分配Keyed State；与定义的最大并行度一样，Key Groups的数量也一样多。在执行过程中，keyed operator的每个并行实例都使用一个或多个Key Groups的键。

**Operator State**

使用Operator State(或非键控状态)，每个算子状态都绑定到一个并行算子实例。Kafka连接器是在Flink中使用算子状态的一个很好的例子。Kafka consumer的每个并行实例都维护主题分区和偏移量的映射作为其算子状态。

当并行性发生变化时，Operator State接口支持在并行算子实例之间重新分配状态。进行这种再分配可以有不同的方案。

## Raw and Managed State（两个基本状态的存在形式） ##

Keyed State 和Operator State以两种形式存在：*managed* and *raw。*

Managed State托管状态由Flink运行时控制的数据结构表示，如内部哈希表或RocksDB。例如“ValueState”、“ListState”等。Flink的运行时对状态进行编码并将它们写入checkpoints。

Raw State是算子保存在它们自己的数据结构中的状态。当检查点时，它们只向检查点写入一个字节序列。Flink不关注数据结构，只看原始字节。

所有datastream函数都可以使用managed state，但是raw state接口只能在实现operators后才能使用。建议使用managed state(而不是raw state)，因为使用managed state Flink可以在并行性发生更改时自动重新分配状态，而且还可以进行更好的内存管理。

如果您的托管状态需要自定义序列化逻辑，请参阅相应的指南（[托管状态的自定义序列化][Link 1]//TODO），以确保将来的兼容性。Flink的默认序列化器不需要特殊处理。

## Using Managed Keyed State（如何使用） ##

managed keyed state接口提供对不同类型状态的访问，这些状态的范围都是当前输入元素的键。这意味着这种状态只能在KeyedStream上使用，KeyedStream可以通过stream.keyBy(…)创建。

现在，我们将首先查看可用状态的不同类型，然后我们将看到如何在程序中使用它们。可用的状态有：

*  `ValueState<T>：`
 *  `ListState<T>：`
 *  `ReducingState<T>：`
 *  `AggregatingState<IN, OUT>：`
 *  `FoldingState<T, ACC>：`
 *  `MapState<UK, UV>：`

所有类型的状态都有一个方法clear()，它为当前活动的键(即输入元素的键)清除状态。

> **注意：**FoldingState和FoldingStateDescriptor已经在Flink 1.4中被弃用，并将在将来完全删除。请使用AggregatingState和AggregatingStateDescriptor。

务必记住，这些状态对象仅用于状态接口。状态不一定存储在内部，但可能存储在磁盘或其他地方。要记住的第二件事是，从状态获得的值取决于输入元素的键。因此，如果涉及的键不同，那么在一次用户函数调用中得到的值可能与在另一次调用中得到的值不同。

获取一个state handle，必须先创建一个`StateDescriptor。`这保存了状态的名称，状态所持有值的类型，用户指定的函数，比如`ReduceFunction。`根据需要检索的状态类型，可以创建ValueStateDescriptor、ListStateDescriptor、ReducingStateDescriptor、FoldingStateDescriptor或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>)`
 *  `FoldingState<T, ACC> getFoldingState(FoldingStateDescriptor<T, ACC>)`
 *  `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(Configuration config) {
            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(0)
            .flatMap(new CountWindowAverage())
            .print();
    
    // the printed output will be (1,4) and (1,5)

这个例子实现了一个简单计数窗口。我们按第一个字段键入元组（在这个例子当中都有一个相同的key1）。该函数将计数和求和存储在ValueState中。一旦计数达到2，它就会下传平均值并清除状态，这样我们就可以从0开始。注意，如果我们在第一个字段中有不同key值的元组，那么这将为每个不同的输入键保留不同的状态值。

### 状态有效期（TTL） ###

任意类型的keyed state都可以制定有效期。如果配置了TTL，并且状态值已经过期，那么将尽最大努力清理存储值，下面将对此进行更详细的讨论。

所有状态集合类型都支持单个条目的TTLs。这意味着列表元素和映射条目的TTL互不影响。

为了使用状态TTL，必须首先构建一个StateTtlConfig配置对象。然后通过传递配置，可以在任何state descriptor中启用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);

配置有几个选项需要考虑：

newBuilder方法的第一个参数是必填的，它是TTL的值。

更新类型在状态TTL刷新时配置(默认情况下是OnCreateAndWrite)：

*  `StateTtlConfig.UpdateType.OnCreateAndWrite` \- 创建和写的时候执行
 *  `StateTtlConfig.UpdateType.OnReadAndWrite` \- 只在读的时候执行

状态可视性配置，如果未清除过期数据(默认情况下是`NeverReturnExpired`)，则是否在读取访问时返回过期值：

*  `StateTtlConfig.StateVisibility.NeverReturnExpired` \- 过期数据永远不返回
 *  `StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp` \- 仍然返回可用数据

这个案例中的`NeverReturnExpired，就像是数据不存在一样，就算是他仍然会被删除。`该选项对于在TTL之后的数据必须严格不可读的用例非常有用，例如处理隐私敏感数据的应用程序。

另一个选项ReturnExpiredIfNotCleanedUp允许在清理之前返回过期状态。

> **注意：**
> 
>  *  状态后端存储最后一次修改的时间戳和用户值，这意味着启用该特性会增加状态存储的消耗。state backend堆在内存中会存储一个Java对象，对象包含用户state对象和一个时间戳。RocksDB state backend会给每个存储值新增8个字节，list entry或map entry。
>  *  TTLs目前仅支持processing time
>  *  之前没有配置过TTL，尝试恢复state，使用TTL需要 descriptor 否则会因为兼容性导致任务失败，并抛出`StateMigrationException。`
>  *  TTL配置不是 check- or savepoints 的 一部分，而是Flink在当前运行的作业中如何处理它的一种方式。
>  *  仅当 user value serializer 可以处理空值时，带有TTL的 map state 才支持空的 user value。如果 serializer 不支持空值，则可以使用 `NullableSerializer `对其进行包装，代价是在序列化形式中增加一个字节。

**Cleanup of Expired State（清除过期状态）**

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

**a - 全量清除**

> **注意：**这意味着默认情况下，如果未读取 expired state，则不会删除它，这可能导致状态不断增长。这可能会在未来的版本中优化。

此外，您可以在获取完整state snapshot时激活清理操作，这将减少其 state 大小。当下，local state 不会被清除，但是在恢复 state 的时候，已经删除的 state 不会被恢复。改选项可以在 `StateTtlConfig 中配置。`

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

此选项不适用于 RocksDB state backend 中的增量 RocksDB state backend。

注意：

*  对于现有的 job，该策略可以在 `StateTtlConfig 中随时启用或停用，例如从 savepoint 后启动。`

`后台清理`

除了从全部的快照中做清除操作以外，下面的选项将在 StateTtlConfig 中配置一个默认的后台清除，前提是他支持backend。

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

如果要对后台的一些特殊的清理操作做更细粒度的控制，请根据以下描述进行配置。目前，heap state backend 依赖于增量清理，而 RocksDB backend 使用压缩过滤器进行后台清理。

**b - 增量清除**

另一种方式是可以增量的触发对某些状态项的清理。触发器可以是来自每个 state access 或/和每个记录处理的回调。如果这个清理策略运行在某个状态上，那么存储后端会在其所有条目上为该状态保留懒加载的一个全局迭代器。每次触发增量清理时，迭代器都会迭代一次。检查遍历的状态项，并清理过期的状态项。

在 `StateTtlConfig 中配置：`

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

这个策略有两个参数，第一个是每次清理触发的检查状态条目数。如果启用，则每次状态访问都将触发它。第二个参数定义是否为每个记录处理额外触发清理。如果启用默认的后台清理，则此策略将为 heap backend 激活，其中包含5个可选项，且每个记录处理不需要清理。

注意：

*  如果没有对状态进行访问或没有处理任何记录，则过期状态将持久存在。
 *  增量清理会增加数据处理的耗时。
 *  目前，增量清理仅为 Heap state backend 的配置。RocksDB设置是没有效果的。
 *  如果使用 heap state backend 进行同步快照，全局迭代器在迭代时保留所有键的副本，因为它的特性并不支持并发修改。
 *  启用此功能将增加内存消耗。异步快照不存在这个问题。
 *  对于现有作业，此清理策略可以在 StateTtlConfig 中随时激活或停用，例如从 savepoint 重新启动后。

**c - RocksDB压缩期间进行清理**

如果使用 RocksDB state backend，另一个清理策略是激活Flink特定的压缩过滤器。RocksDB 定期运行异步压缩来合并状态更新和减少存储。Flink压缩过滤器使用TTL检查状态项的过期时间戳，不包括过期值。

默认情况下此功能是禁用的。必须首先为 RocksDB backend 后端激活它，方法是设置Flink配置选项 `state.backend.rocksdb.ttl.compaction.filter.enabled`，或者在为作业创建自定义 RocksDB state backend 时调用 `RocksDBStateBackend::enableTtlCompactionFilter`。

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

RocksDB压缩过滤器将在每次处理一定数量的状态项之后，从Flink查询用于检查当前过期的时间戳。你可以修改并将自定义的值传给 `StateTtlConfig.newBuilder(...).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries)。`更频繁的刷新时间戳可以提高清理速度，但会降低压缩性能，因为它使用本地代码的JNI调用。如果启用默认的后台清理，那么RocksDB后端将激活此策略，每次处理1000条记录时将查询当前时间戳。

通过激活FlinkCompactionFilter的调试级别，您可以从RocksDB filter的本地代码中激活调试日志：

log4j.logger.org.rocksdb.FlinkCompactionFilter=DEBUG

> 注意：
> 
>  *  在压缩过程中调用TTL过滤器会减慢它的处理速度。TTL过滤器必须解析上次访问的时间戳，并检查每个正在压缩的键的每个存储状态条目的过期时间。对于集合状态类型(列表或映射)，每个存储的元素也调用该检查策略。
>  *  对于元素序列化后长度不固定的列表状态，TTL 过滤器需要在每次 JNI 调用过程中，额外调用 Flink 的 java 序列化器， 从而确定下一个未过期数据的位置。
>  *  对于正在运行的 jobs，该策略可以在 `StateTtlConfig 配置中随时启用或停用，例如：重启 `savepoint 以后。

**Scala DataStream API中的State**

除了上面描述的接口之外，Scala API还为有状态 map()或 flatMap()函数提供了快捷方式，这些函数在 KeyedStream 上只有一个ValueState。用户函数在 `Option `中获取 ValueState 的当前值，并且必须返回一个更新后的值，该值将用于更新状态。

val stream: DataStream[(String, Int)] = ...
    
    val counts: DataStream[(String, Int)] = stream
      .keyBy(_._1)
      .mapWithState((in: (String, Int), count: Option[Int]) =>
        count match {
          case Some(c) => ( (in._1, c), Some(c + in._2) )
          case None => ( (in._1, 0), Some(in._2) )
        })

## 使用Managed Operator State ##

使用 managed operator state 需要实现 `CheckpointedFunction 接口或者 ListCheckpointed<T extends Serializable> 接口。`

**1-CheckpointedFunction**

CheckpointedFunction接口通过不同的重新分配方案提供对 non-keyed state 的访问。它需要实现两种方法：

void snapshotState(FunctionSnapshotContext context) throws Exception;
    
    void initializeState(FunctionInitializationContext context) throws Exception;

无论何时在 checkpoint 之后，都会调用 snapshotState()。每当用户初始化自定义的函数时，都会调用对应的 initializeState()，无论是在函数第一次初始化时，还是在函数 checkpoint 恢复数据时。因此，initializeState()不仅是初始化不同类型状态的地方，而且是包含状态恢复逻辑的地方。

当前支持列表样式的 managed operator state，状态应该是一个可序列化对象列表，彼此独立，因此在重新缩放时可以重新分配。换句话说，这些对象是可以重新分布 non-keyed state 的最细粒度。根据状态访问方法，定义了以下重分发方案：

*  **Even-split redistribution：**每个算子返回一个状态元素列表。整个状态在逻辑上是所有列表的连接。在恢复/重新分发中，列表被平均地划分为尽可能多的子列表，只要有并行算子即可。每个算子都有一个子列表，它可以是空的，也可以包含一个或多个元素。例如，如果在 parallelism 1中，算子的 checkpoint 状态包含元素element1和element2，当将并行度增加到2时，element1可能最终出现在算子0中，而element2将出现在算子1中。
 *  **Union redistribution：**每个算子返回一个状态元素列表。整个状态在逻辑上是所有列表的连接。在恢复/重新分发时，每个操作符都获得状态元素的完整列表。

下面是一个有状态 SinkFunction 的例子，它使用 CheckpointedFunction 在将他们发出之前会缓存元素。下面的例子是一个基本的 even-split redistribution list state：

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.clear();
            for (Tuple2<String, Integer> element : bufferedElements) {
                checkpointedState.add(element);
            }
        }
    
        @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类型的容器，其中 non-keyed state 对象将在 checkpointing 时存储。

注意：状态是如何初始化的，类似于 keyed state，使用一个 `StateDescriptor`，其中包含状态名和关于状态所持有值的类型的信息：

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

状态访问方法的命名约定，包含其重新分布模式及其状态结构。例如，使用带有 union redistribution 的状态列表，使用 `getUnionListState(descriptor) 访问状态。`如果方法名不包含 redistribution pattern，例如getListState(`descriptor`)，它仅仅意味着将使用基本的 even-split redistribution scheme。

初始化 container 之后，我们使用上下文的 `isRestored`()方法来检查在发生故障后是否正在恢复。如果返回结果为 true，表示正在进行恢复操作，则应用恢复逻辑。

如修改后的 BufferingSink 代码所示，在状态初始化期间恢复的 `ListState `保存在一个类变量中，供 snapshotState()以后使用。在那里，ListState清除了前一个 checkpoint 包含的所有对象，然后填充我们想要 checkpoint 的新对象。

另外，keyed state 也可以在 initializeState()方法中初始化。这可以使用提供的 FunctionInitializationContext 来完成。

**2-ListCheckpointed**

`ListCheckpointed` 接口是 `CheckpointedFunction `的精简版，它只支持 list-style state 和恢复时的 even-split redistribution scheme。它还需要实现两种方法：

List<T> snapshotState(long checkpointId, long timestamp) throws Exception;
    
    void restoreState(List<T> state) throws Exception;

在 `snapshotState`()上，算子应该将对象列表下发给 checkpoint，而 `restoreState `必须在恢复时处理列表。如果状态不可重分区，则始终可以在 `snapshotState`()中返回 `Collections.singletonList(MY_STATE)`。

### **Stateful Source Functions（带状态的Source Functions）** ###

带状态的数据源相对与其他算子来说需要注意的更多。为了对状态和输出集合的原子性进行更新(对于故障/恢复时的 exactly-once 语义来说，这是必需的)，用户需要从源上下文获取一个锁。

public static class CounterSource
            extends RichParallelSourceFunction<Long>
            implements ListCheckpointed<Long> {
    
        /**  current offset for exactly once semantics */
        private Long offset = 0L;
    
        /** flag for job cancellation */
        private volatile boolean isRunning = true;
    
        @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 List<Long> snapshotState(long checkpointId, long checkpointTimestamp) {
            return Collections.singletonList(offset);
        }
    
        @Override
        public void restoreState(List<Long> state) {
            for (Long s : state)
                offset = s;
        }
    }

当一个 checkpoint 被Flink ACK时，一些算子可能需要这些信息来与外部世界通信。在本例中，请参见org.apache.flink.runtime.state.CheckpointListener 接口。

[Link 1]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/stream/state/custom_serialization.html