聊聊Flink：Flink的状态管理

一、Flink的状态是什么？

我们知道，Flink的一个算子可能会有多个子任务，每个子任务可能分布在不同的实例（即slot）上，我们可以把Flink的状态理解为某个算子的子任务在其当前实例上的一个变量，该变量记录了流过当前实例算子的历史记录产生的结果。当新数据记录流入时，我们需要结合该结果（即状态，State）来进行计算。

实际上，Flink的状态是由算子的子任务来创建和管理的。一个状态的更新和获取的流程如下图所示，一个算子子任务接收输入流，获取对应的状态，根据新的计算结果更新状态。一个简单的例子是对一个时间窗口内流入的某个整数字段进行求和，那么当算子子任务接收到新元素时，会获取已经存储在状态中的数值（历史记录的求和结果），然后将当前输入加到状态上，并将状态数据更新。

在这里插入图片描述
Flink应用程序的状态访问都在本地进行，这样有助于提高吞吐量和降低延迟。通常情况下，Flink应用程序都是将状态存储在JVM堆内存中，但如果状态数据太大，也可以选择将其以结构化数据格式存储在高速磁盘中。

在这里插入图片描述
通过状态快照，Flink能够提供可容错的、精确一次的计算语义。Flink应用程序在执行时会获取并存储分布式Pipeline（流处理管道）中整体的状态，它会将数据源中消费数据的偏移量记录下来，并将整个作业图中算子获取到该数据（记录的偏移量对应的数据）时的状态记录并存储下来。当发生故障时，Flink作业会恢复上次存储的状态，重置数据源从状态中记录的上次消费的偏移量，开始重新进行消费处理。而且状态快照在执行时会异步获取状态并存储，并不会阻塞正在进行的数据处理逻辑。

总结来说，Flink状态管理的主要特性如下：

本地性：Flink状态是存储在使用它的机器本地的，并且可以内存访问速度来获取。
持久性：Flink状态是容错的，例如它可以自动按一定的时间间隔产生快照，并且在任务失败后进行恢复。
纵向可扩展性：Flink状态可以存储在集成的RocksDB（一种KV型数据库）实例中，这种方式下可以通过增加本地磁盘来扩展空间。
横向可扩展性：Flink状态可以随着集群的扩／缩容重新分布。
可查询性：Flink状态可以通过使用状态查询API从外部进行查询。

Flink提供了不同的状态机制，用于指定状态的存储方式和存储位置。根据数据集是否按照Key进行分区，将状态分为Keyed State和Operator State(Non-Keyed State)两种类型。

二、Keyed State 与 Operator State

2.1 Keyed State

Keyed State在通过keyBy()分组的KeyedStream上使用，对每个Key的数据进行状态存储和管理，状态是跟每个Key绑定的，即每个Key对应一个状态对象。根据状态数据的类型不同，Flink中定义了多种状态对象，用于存储状态数据，以适应不同的计算场景。

通过keyBy()会将数据流进行状态分区，Keyed State被进一步组织成所谓的Key Groups，一个Key Groups包含多个Key的状态。Key Groups是Flink可以重新分配Keyed State的原子单位，Key Groups的数量与定义的最大并行度相同。在执行期间，算子的每个并行实例处理一个Key Groups，如下图所示。

在这里插入图片描述
Keyed State支持的状态数据类型如下：

ValueState：保存一个具体的值，可以对其更新和查询。该值对应当前输入数据的Key，相当于Key的聚合结果，即每个key都可能对应一个值，而每个值都存储在ValueState对象中。ValueState对象中的值可以通过update(T)方法进行更新，通过T value()方法进行获取。例如统计用户ID对应的交易次数，每次用户交易都会在状态值上进行更新。
ListState：保存一个数据列表。可以通过add(T)或者addAll(List)往这个列表中追加数据；通过get()获得一个Iterable，以便能够遍历整个列表；还可以通过update(List)覆盖当前的列表。例如定义ListState存储用户经常访问的IP地址。
ReducingState：保存一个单值。使用add(T)增加元素，每次调用add(T)方法添加元素时，都会调用用户定义的ReduceFunction，将结果合并为一个元素，并更新到状态中。当要对状态进行聚合计算时，可以使用ReducingState。
AggregatingState<IN, OUT>：保留一个单值。和ReducingState不同的是，聚合类型可能与添加到状态的元素的类型不同。使用add(IN)添加元素，并与已有的元素使用AggregateFunction进行聚合。
MapState<UK, UV>：使用Map存储键值列表。使用put(UK，UV)或者putAll(Map<UK，UV>)添加键值对到状态中，使用get(UK)根据Key查询。

例如，使用ValueState进行简单的计数，流数据中的相同Key一旦出现次数达到2，则将其平均值发送到下游，并清除状态重新开始，代码如下：

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(value -> value.f0)
        .flatMap(new CountWindowAverage())
        .print();

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

2.2 Operator State

与Keyed State不同，Operator State与一个特定算子的一个实例绑定，和数据元素中的Key无关，每个算子子任务管理自己的Operator State，每个算子子任务上的数据流共享同一个状态，可以访问和修改该状态。Kafka连接器是在Flink中使用Operator State的一个很好的例子。Kafka消费者的每个并行实例维护一个主题分区和偏移量的映射作为它的Operator State。

当并行性发生改变时，Operator State接口支持在并行算子实例之间重新分配状态，并且有不同的重新分配方案。Operator State是一种特殊类型的状态，主要用于Source或Sink算子，用来保存流入数据的偏移量或对输出数据做缓存，以保证Flink应用的Exactly-Once语义。

Operator State支持的状态数据类型为ListState。ListState以一个列表的形式存储状态数据，以适应横向扩展时状态重分布的问题。ListState存储的列表数据是相互独立的状态项的集合，在算子并行性发生改变时，这些状态项可以在算子实例之间重新分配。

广播状态(Broadcast State)是Operator State的一种特殊类型。引入广播状态是为了支持需要将一个流的记录广播到所有下游任务的场景，这些记录用于在所有子任务之间保持相同的状态。然后可以在处理第二个流的记录时访问此状态。可以想象一个低吞吐量流，其中包含一组规则，我们要使用这组规则针对来自另一个流的所有元素进行评估。它仅适用于具有广播流和非广播流作为输入的特定算子，并且这样的算子可以具有名称不同的多个广播状态。

我们可以通过实现CheckpointedFunction接口来使用Operator State。该接口是有状态转换函数的核心接口，即跨单个流记录维护状态的函数。虽然有更多轻量级的接口作为各种状态的快捷方式，但该接口在管理Keyed State和Operator State方面提供了最大的灵活性。

在这里插入图片描述
Operator State支持的数据结构如下：

列表状态（List state）：将状态表示为一组数据的列表。
联合列表状态（Union list state）：也将状态表示为数据的列表，它与常规列表状态的区别在于，状态缩放时状态该如何分配。ListState是将整个状态列表按照round-ribon的模式均匀分布到各个算子子任务上，而Union list state按照广播的模式，将所有状态合并，再分发给每个实例的子任务上。
广播状态（Broadcast state）：如果一个算子有多项任务，而它的每项任务状态又都相同，那么这种特殊情况最适合应用广播状态。广播状态是固定维护在堆内存中的，不会写入文件系统或者RocksDB。广播流一侧修改广播状态的键值之后，数据流一侧就可以立即感知到变化。在开发过程中，如果遇到下发/广播配置、规则等低吞吐事件流到下游所有task时，就可以使用Broadcast state的特性。

CheckpointedFunction接口的源码如下：

在这里插入图片描述
当检查点（Checkpoint）获取转换函数的状态快照（Snapshot）时，将调用snapshotState(FunctionSnapshotContext)。在这个方法中，函数通常确保检查点数据结构（在初始化阶段获得的）是最新的，以便进行快照。给定的FunctionSnapshotContext（快照上下文）提供对检查点元数据的访问。当算子的子任务初始化（实例化）时，initializeState(FunctionInitializationContext)被调用。子任务初始化包括第一次自定义函数初始化和从之前的Checkpoint恢复，因此初始化有两种应用场景：

Flink作业第一次执行时，状态数据被初始化为一个默认值。
作业已经将状态数据保存到外部存储，当Flink应用重启时，通过这个方法读取外部存储的状态数据并填充到当前本地状态中。

下面的代码通过一个简单的示例演示使用SinkFunction输出数据到外部系统，并在CheckpointedFunction中进行数据缓存，然后统一发送到下游。当作业重启的时候，对状态数据进行恢复并重新分配。

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<>();
    }

    // sink的核心处理逻辑，将给定的值写入sink。每个记录都会调用此函数。
    @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) {
                // 输出到外部系统（需要自行实现）
            }
            // 清空本地缓存
            bufferedElements.clear();
        }
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        // 清除状态
        checkpointedState.clear();
        // 将本地状态添加到ListState
        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>>() {}));
        
        // 创建ListState，每个ListState都使用唯一的名称。
        checkpointedState = context.getOperatorStateStore().getListState(descriptor);

        // 如果从前一个执行的快照恢复状态，则返回true（例如作业重启的情况）
        if (context.isRestored()) {
            for (Tuple2<String, Integer> element : checkpointedState.get()) {
                bufferedElements.add(element);
            }
        }
    }
}

2.3 Keyed State 与 Operator State 的区别

	Keyed State	Operator State
适用算子类型	只适用于KeyedStream上的算子	可以用于所有算子
状态分配	每个Key对应一个状态	一个算子子任务对应一个状态
创建和访问方式	重写对应的算子Rich Function，通过里面的RuntimeContext访问	实现CheckpointedFunction等接口
状态缩放	状态随着Key自动在多个算子任务上迁移	有多种状态重新分配方式
支持的数据结构	ValueState、ListState、MapState、Reducing state、Aggregating State	List state、Union list state、Broadcast state

三、Keyed State

3.1 KeyedState之ValueState
ValueState[T]是单一变量的状态，T是某种具体的数据类型，比如Double、String，或我们自己定义的复杂数据结构。我们可以使用value()方法获取状态，使用update(T value)更新状态。

需求：当接收到的相同 key 的元素个数等于 3 个，就计算这些元素的 value 的平均值。

（1）继承算子的RichFunction，创建状态并编写业务逻辑。

public class CountWindowAverageWithValueState extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Double>> {

    /**
     * 用以保存每个 key 出现的次数，以及这个 key 对应的 value 的总值
     * 1. ValueState 保存的是对应的一个 key 的一个状态值
     */
    private ValueState<Tuple2<Long, Long>> countAndSum;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 注册状态
        ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
                new ValueStateDescriptor<>(
                        "average",  // 状态的名字
                        Types.TUPLE(Types.LONG, Types.LONG)); // 状态存储的数据类型，防止类型擦除
        countAndSum = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void flatMap(
            Tuple2<Long, Long> element,
            Collector<Tuple2<Long, Double>> out) throws Exception {
        // 拿到当前的 key 的状态值
        Tuple2<Long, Long> currentState = countAndSum.value();
        // 如果状态值还没有初始化，则初始化
        if (currentState == null) {
            currentState = Tuple2.of(0L, 0L);
        }
        // 更新状态值中的元素的个数
        currentState.f0 += 1;
        // 更新状态值中的总值
        currentState.f1 += element.f1;
        // 更新状态
        countAndSum.update(currentState);
        // 判断，如果当前的 key 出现了 3 次，则需要计算平均值，并且输出
        if (currentState.f0 >= 3) {
            double avg = (double)currentState.f1 / currentState.f0;
            // 输出 key 及其对应的平均值
            out.collect(Tuple2.of(element.f0, avg));
            //  清空状态值
            countAndSum.clear();
        }
    }
}

（2）Main方法

/**
 * 需求：当接收到的相同 key 的元素个数等于 3 个,就计算这些元素的 value 的平均值。
 */
public class TestKeyedStateMain {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Tuple2<Long, Long>> dataStreamSource =
                env.fromElements(Tuple2.of(1L, 3L), Tuple2.of(1L, 5L),
                        Tuple2.of(1L, 7L),
                        Tuple2.of(2L, 4L), Tuple2.of(2L, 2L), Tuple2.of(2L,
                                5L));
        // 输出：
        //(1,5.0)
        //(2,3.6666666666666665)
        dataStreamSource
                .keyBy(0)
                .flatMap(new CountWindowAverageWithValueState())
                .print();
        env.execute();
    }
}

（3）输出

11> (1,5.0)
16> (2,3.6666666666666665)

3.2 KeyedState之ListState
ListState[T]存储了一个由T类型数据组成的列表。我们可以使用add(IN var1)或addAll(List var1)向状态中添加元素，使用OUT get()获取整个列表，使用update(List var1)来更新列表，新的列表将替换旧的列表。

和3.1类似，main方法换一下CountWindowAverageWithListState类就可以了。

public class CountWindowAverageWithListState extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Double>> {
    //1. ListState 保存的是对应的一个 key 的出现的所有的元素
    private ListState<Tuple2<Long, Long>> elementsByKey;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 注册状态
        ListStateDescriptor<Tuple2<Long, Long>> descriptor =
            new ListStateDescriptor<>(
                "average",  // 状态的名字
                Types.TUPLE(Types.LONG, Types.LONG)); // 状态存储的数据类型
        elementsByKey = getRuntimeContext().getListState(descriptor);
    }
    @Override
    public void flatMap(Tuple2<Long, Long> element,
                        Collector<Tuple2<Long, Double>> out) throws Exception {
        // 拿到当前的 key 的状态值
        Iterable<Tuple2<Long, Long>> currentState = elementsByKey.get();
        // 如果状态值还没有初始化，则初始化
        if (currentState == null) {
            elementsByKey.addAll(Collections.emptyList());
        }
        // 更新状态
        elementsByKey.add(element);
        // 判断，如果当前的 key 出现了 3 次，则需要计算平均值，并且输出
        List<Tuple2<Long, Long>> allElements = new ArrayList<>((Collection<? extends Tuple2<Long, Long>>) elementsByKey.get());
        if (allElements.size() >= 3) {
            long count = 0;
            long sum = 0;
            for (Tuple2<Long, Long> ele : allElements) {
                count++;
                sum += ele.f1;
            }
            double avg = (double) sum / count;
            out.collect(Tuple2.of(element.f0, avg));
            // 清除状态
            elementsByKey.clear();
        }
    }
}

3.3 KeyedState之MapState
MapState<UK, UV>存储一个Key-Value map，其功能与Java的Map几乎相同。UV get(UK var1)可以获取某个key下的value，void put(UK var1, UV var2)可以对某个key设置value，boolean contains(UK var1)判断某个key是否存在，void remove(UK var1)删除某个key以及对应的value，Iterable<Entry<UK, UV>> entries()返回MapState中所有的元素，Iterator<Entry<UK, UV>> iterator()返回一个迭代器。需要注意的是，MapState中的key和Keyed State的key不是同一个key。

public class CountWindowAverageWithMapState extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Double>> {
    // managed keyed state
    //1. MapState ：key 是一个唯一的值，value 是接收到的相同的 key 对应的 value 的值
    private MapState<String, Long> mapState;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 注册状态
        MapStateDescriptor<String, Long> descriptor =
            new MapStateDescriptor<>(
                "average",  // 状态的名字
                String.class, Long.class); // 状态存储的数据类型
        mapState = getRuntimeContext().getMapState(descriptor);
    }
    @Override
    public void flatMap(Tuple2<Long, Long> element,
                        Collector<Tuple2<Long, Double>> out) throws Exception {
        mapState.put(UUID.randomUUID().toString(), element.f1);
        // 判断，如果当前的 key 出现了 3 次，则需要计算平均值，并且输出
        List<Long> allElements = new ArrayList<>((Collection<? extends Long>) mapState.values());

        if (allElements.size() >= 3) {
            long count = 0;
            long sum = 0;
            for (Long ele : allElements) {
                count++;
                sum += ele;
            }
            double avg = (double) sum / count;
            out.collect(Tuple2.of(element.f0, avg));
            // 清除状态
            mapState.clear();
        }
    }
}

3.4 KeyedState之ReducingState

ReducingState[T]和AggregatingState[IN, OUT]与ListState[T]同属于MergingState[T]。与ListState[T]不同的是，ReducingState[T]只有一个元素，而不是一个列表。它的原理是新元素通过void add(IN var1)加入后，与已有的状态元素使用ReduceFunction合并为一个元素，并更新到状态里。AggregatingState[IN, OUT]与ReducingState[T]类似，也只有一个元素，只不过AggregatingState[IN, OUT]的输入和输出类型可以不一样。ReducingState[T]和AggregatingState[IN, OUT]与窗口上进行ReduceFunction和AggregateFunction很像，都是将新元素与已有元素做聚合。

需求：求接收到的相同key的value的sum。

public class SumFunction extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
    // 用于保存每一个 key 对应的 value 的总值
    private ReducingState<Long> sumState;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 注册状态
        // 聚合函数
        ReducingStateDescriptor<Long> descriptor =
            new ReducingStateDescriptor<>(
                "sum",  // 状态的名字
                (ReduceFunction<Long>) Long::sum, Long.class); // 状态存储的数据类型
        sumState = getRuntimeContext().getReducingState(descriptor);
    }
    @Override
    public void flatMap(Tuple2<Long, Long> element,
                        Collector<Tuple2<Long, Long>> out) throws Exception {
        // 将数据放到状态中
        sumState.add(element.f1);

        out.collect(Tuple2.of(element.f0, sumState.get()));
    }
}

（2）Main类

将3.1的Main方法中flatMap的Function替换为SumFunction

（3）输出

16> (2,4)
11> (1,3)
16> (2,6)
11> (1,8)
16> (2,11)
11> (1,15)

3.5 KeyedState之AggregatingState
需求：求接收到的相同 key 的value显示出来。

（1）继承算子的RichFunction，创建状态并编写业务逻辑。

public class ContainsValueFunction extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, String>> {
    private AggregatingState<Long, String> totalStr;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 注册状态
        AggregatingStateDescriptor<Long, String, String> descriptor =
            new AggregatingStateDescriptor<>(
                "totalStr",  // 状态的名字
                new AggregateFunction<Long, String, String>() {
                    @Override
                    public String createAccumulator() {
                        return null;
                    }

                    @Override
                    public String add(Long value, String accumulator) {
                        if (StringUtils.isBlank(accumulator)) {
                            return String.valueOf(value);
                        }
                        return accumulator + " and " + value;
                    }

                    @Override
                    public String getResult(String accumulator) {
                        return accumulator;
                    }

                    @Override
                    public String merge(String a, String b) {
                        return null;
                    }
                }, String.class); // 状态存储的数据类型
        totalStr = getRuntimeContext().getAggregatingState(descriptor);
    }

    @Override
    public void flatMap(Tuple2<Long, Long> element,
                        Collector<Tuple2<Long, String>> out) throws Exception {
        totalStr.add(element.f1);
        out.collect(Tuple2.of(element.f0, totalStr.get()));
    }
}

（2）Main方法

将3.1的Main方法中flatMap的Function替换为ContainsValueFunction。

（3）输出

11> (1,3)
16> (2,4)
11> (1,3 and 5)
16> (2,4 and 2)
11> (1,3 and 5 and 7)
16> (2,4 and 2 and 5)

四、Operator State

4.1 OperatorState之ListState

状态从本质上来说，是Flink算子子任务的一种本地数据，为了保证数据可恢复性，使用Checkpoint机制来将状态数据持久化输出到存储空间上。状态相关的主要逻辑有两项：

将算子子任务本地内存数据在Checkpoint时snapshot写入存储；
初始化或重启应用时，以一定的逻辑从存储中读出并变为算子子任务的本地内存数据。

Keyed State对这两项内容做了更完善的封装，开发者可以开箱即用。对于Operator State来说，每个算子子任务管理自己的Operator State，或者说每个算子子任务上的数据流共享同一个状态，可以访问和修改该状态。Flink的算子子任务上的数据在程序重启、横向伸缩等场景下不能保证百分百的一致性。换句话说，重启Flink应用后，某个数据流元素不一定会和上次一样，还能流入该算子子任务上。因此，我们需要根据自己的业务场景来设计snapshot和restore的逻辑。为了实现这两个步骤，Flink提供了最为基础的CheckpointedFunction接口类。

public interface CheckpointedFunction {
  
  // Checkpoint时会调用这个方法，我们要实现具体的snapshot逻辑，比如将哪些本地状态持久化
  void snapshotState(FunctionSnapshotContext context) throws Exception;
  // 初始化时会调用这个方法，向本地状态中填充数据
  void initializeState(FunctionInitializationContext context) throws Exception;
}

在Flink的Checkpoint机制下，当一次snapshot触发后，snapshotState会被调用，将本地状态持久化到存储空间上。这里我们可以先不用关心snapshot是如何被触发的，暂时理解成snapshot是自动触发的，后续文章会介绍Flink的Checkpoint机制。

initializeState在算子子任务初始化时被调用，初始化包括两种场景：

整个Flink作业第一次执行，状态数据被初始化为一个默认值；
Flink作业重启，之前的作业已经将状态输出到存储，通过这个方法将存储上的状态读出并填充到这个本地状态中。

目前Operator State主要有三种，其中ListState和UnionListState在数据结构上都是一种ListState，还有一种BroadcastState。这里我们主要介绍ListState这种列表形式的状态。这种状态以一个列表的形式序列化并存储，以适应横向扩展时状态重分布的问题。每个算子子任务有零到多个状态S，组成一个列表ListState[S]。各个算子子任务将自己状态列表的snapshot到存储，整个状态逻辑上可以理解成是将这些列表连接到一起，组成了一个包含所有状态的大列表。当作业重启或横向扩展时，我们需要将这个包含所有状态的列表重新分布到各个算子子任务上。

ListState和UnionListState的区别在于：

ListState是将整个状态列表按照round-ribon的模式均匀分布到各个算子子任务上，每个算子子任务得到的是整个列表的子集；
UnionListState按照广播的模式，将整个列表发送给每个算子子任务。

Operator State的实际应用场景不如Keyed State多，它经常被用在Source或Sink等算子上，用来保存流入数据的偏移量或对输出数据做缓存，以保证Flink应用的Exactly-Once语义。这里我们来看一个Flink官方提供的Sink案例以了解CheckpointedFunction的工作原理。

需求: 每两条数据打印一次结果 1000

（1）实现SinkFunction和CheckpointedFunction

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

    // 用于缓存结果数据的
    private List<Tuple2<String, Integer>> bufferElements;
    // 表示内存中数据的大小阈值
    private int threshold;
    // 用于保存内存中的状态信息
    private ListState<Tuple2<String, Integer>> checkpointState;
    // StateBackend
    // checkpoint
    public CustomSink(int threshold) {
        this.threshold = threshold;
        this.bufferElements = new ArrayList<>();
    }

    // Sink的核心处理逻辑，将上游数据value输出到外部系统
    @Override
    public void invoke(Tuple2<String, Integer> value, Context context) throws Exception {
        // 可以将接收到的每一条数据保存到任何的存储系统中
        bufferElements.add(value);
        if (bufferElements.size() == threshold) {
            // send it to the sink
            // 这里简单打印
            System.out.println("自定义格式：" + bufferElements);
            // 清空本地缓存
            bufferElements.clear();
        }
    }

    // 重写CheckpointedFunction中的snapshotState
    // 将本地缓存snapshot保存到存储上
    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception
    {
        // 将之前的Checkpoint清理
        checkpointState.clear();
        // 将最新的数据写到状态中
        for (Tuple2<String, Integer> ele : bufferElements) {
            checkpointState.add(ele);
        }
    }

    // 重写CheckpointedFunction中的initializeState
    // 初始化状态：用于在程序恢复的时候从状态中恢复数据到内存
    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        // 注册ListStateDescriptor
        ListStateDescriptor<Tuple2<String, Integer>> descriptor =
            new ListStateDescriptor<>(
                "bufferd -elements",
                TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {}));
        // 从FunctionInitializationContext中获取OperatorStateStore，进而获取ListState
        checkpointState = context.getOperatorStateStore().getListState(descriptor);
        // 如果是作业重启，读取存储中的状态数据并填充到本地缓存中
        if (context.isRestored()) {
            for (Tuple2<String, Integer> ele : checkpointState.get()) {
                bufferElements.add(ele);
            }
        }
    }
}

（2）Main方法

/**
 * 需求: 每两条数据打印一次结果 1000
 */
public class TestOperatorStateMain {
    public static void main(String[] args) throws  Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Tuple2<String, Integer>> dataStreamSource =
            env.fromElements(Tuple2.of("Spark", 3), Tuple2.of("Hadoop", 5),
                Tuple2.of("Hadoop", 7),
                Tuple2.of("Spark", 4));

        dataStreamSource.addSink(new CustomSink(2)).setParallelism(1);
        env.execute("TestStatefulApi");
    }
}

（3）输出

自定义格式：[(Spark,3), (Hadoop,5)]
自定义格式：[(Hadoop,7), (Spark,4)]

上面的代码在输出到Sink之前，先将数据放在本地缓存中，并定期进行snapshot，这实现了批量输出的功能，批量输出能够减少网络等开销。同时，程序能够保证数据一定会输出外部系统，因为即使程序崩溃，状态中存储着还未输出的数据，下次启动后还会将这些未输出数据读取到内存，继续输出到外部系统。

注册和使用Operator State的代码和Keyed State相似，也是先注册一个StateDescriptor，并指定状态名字和数据类型，然后从FunctionInitializationContext中获取OperatorStateStore，进而获取ListState。如果是UnionListState，那么代码改为：context.getOperatorStateStore.getUnionListState。

状态的初始化逻辑中，我们用context.isRestored来判断是否为作业重启，这样可以从之前的Checkpoint中恢复并写到本地缓存中。

4.2 OperatorState之BroadCastState

广播状态是固定维护在堆内存中的，不会写入文件系统或者RocksDB。

下面我们通过BroadCastState控制程序的打印输出为例进行介绍。

（1）定义普通数据流，消费数据

DataStreamSource<String> dataStreamSource = env.socketTextStream("localhost", 9999);

（2）定义广播流，用于广播规则，从而控制程序打印输出

DataStreamSource<String> broadStreamSource = env.socketTextStream("localhost", 8888);

（3）解析广播流中的数据，解析为二元组

DataStream<Tuple2<String, String>> broadStream =
    broadStreamSource.map(new MapFunction<String, Tuple2<String, String>>() {
        @Override
        public Tuple2<String, String> map(String s) throws Exception {
            String[] strings = s.split(" ");
            return Tuple2.of(strings[0], (strings[1]));
        }
    });

（4）定义需要广播的状态类型，只支持

MapStateDescriptor<String, String> descriptor = new
    MapStateDescriptor<String, String>(
    "ControlStream",
    String.class,
    String.class
);

（5）用解析后的广播流将状态广播出去，从而生成BroadcastStream

BroadcastStream<Tuple2<String, String>> broadcastStream = broadStream.broadcast(descriptor);

（6）通过connect连接两个流，用process分别处理两个流中的数据。连接流时分为两种情况：

noKeyedStream.connect(BroadcastStream).process(new BroadcastProcessFunction<>(…))：非 KeyedStream 连接 BroadcastStream 的，只能使用 BroadcastProcessFunction 函数处理连接逻辑。
KeyedStream.connect(BroadcastStream).process(new KeyedBroadcastProcessFunction<>(…))：KeyedStream 连接 BroadcastStream 的，只能使用 KeyedBroadcastProcessFunction 函数处理连接逻辑。

KeyedBroadcastProcessFunction 比 BroadcastProcessFunction 多了计时器服务和获取当前 key 接口，当然，这两个功能不一定能用到。

我们这里使用的是 BroadcastProcessFunction<IN1, IN2, OUT>，这三个泛型翻译分别代表：

IN1:数据流(即非广播流)的元素类型
IN2:广播流的元素类型
OUT:两个流连接完成后，输出流的元素类型。

BroadcastProcessFunction中定义了两个函数用于处理具体的连接逻辑和业务逻辑。因此主要需要实现以下两个函数：

public abstract void processBroadcastElement(final IN2 value, final Context ctx, final Collector<OUT> out) throws Exception;

这里处理广播流的数据，将广播流数据保存到 BroadcastState 中。value 是广播流中的一个元素；ctx 是上下文，提供 BroadcastState 和修改方法；out 是输出流收集器。

public abstract void processElement(final IN1 value, final ReadOnlyContext ctx, final Collector<OUT> out) throws Exception;

这个函数处理数据流的数据，这里之只能获取到 ReadOnlyBroadcastState，因为 Flink 不允许在这里修改 BroadcastState 的状态。value 是数据流中的一个元素；ctx 是上下文，可以提供上下文环境和只读的 BroadcastState；out 是输出流收集器。

注意：KeyedBroadcastProcessFunction中的ReadOnlyContext多了计时器服务和获取当前 key 接口

整的代码如下：

需求：通过BroadCastState控制程序的打印输出

/**
 * 数据流：
 * i love flink
 * 广播流：
 * key  flink  -> 代表数据流里面，只要包含flink的单词才会被打印出来。
 */
public class TestBroadcastState {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1. 定义普通数据流，消费数据
        DataStreamSource<String> dataStreamSource = env.socketTextStream("localhost", 9999);
        // 2. 定义广播流，用于广播规则，从而控制程序打印输出
        DataStreamSource<String> broadStreamSource = env.socketTextStream("localhost", 8888);
        // 3. 解析广播流中的数据成二元组
        DataStream<Tuple2<String, String>> broadStream =
            broadStreamSource.map((MapFunction<String, Tuple2<String, String>>) s -> {
                String[] strings = s.split(" ");
                return Tuple2.of(strings[0], (strings[1]));
            });
        //4. 定义需要广播的状态类型，只支持MapState
        MapStateDescriptor<String, String> descriptor = new
            MapStateDescriptor<>(
            "ControlStream",
            String.class,
            String.class
        );
        //5. 用解析后的广播流将状态广播出去，从而生成BroadcastStream
        BroadcastStream<Tuple2<String, String>> broadcastStream = broadStream.broadcast(descriptor);
        //6. 通过connect连接两个流，用process分别处理两个流中的数据
        dataStreamSource
            .connect(broadcastStream)
            .process(new KeyWordsCheckProcessor())
            .print();
        env.execute();
    }

    private static class KeyWordsCheckProcessor extends BroadcastProcessFunction<String, Tuple2<String, String>, String> {
        MapStateDescriptor<String, String> descriptor =
            new MapStateDescriptor<>(
                "ControlStream",
                String.class,
                String.class
            );
        @Override
        public void processBroadcastElement(Tuple2<String, String> value, Context ctx, Collector<String> out) throws Exception {
            // 将接收到的控制数据放到 broadcast state 中
            ctx.getBroadcastState(descriptor).put(value.f0, value.f1);
            // 打印控制信息
            System.out.println(Thread.currentThread().getName() + " 接收到控制信息 ：" + value);
        }

        @Override
        public void processElement(String value, ReadOnlyContext ctx, Collector<String> out) throws Exception {
            // 从 broadcast state 中拿到控制信息
            String keywords = ctx.getBroadcastState(descriptor).get("key");
            // 获取符合条件的单词
            if (value.contains(keywords)) {
                out.collect(value);
            }
        }
    }
}

五、状态后端State backend

5.1 状态后端

Flink将Checkpoint快照的存储位置称为状态后端(State Backend)。状态后端有两种实现：一种基于RocksDB将工作状态保存在磁盘上，使用RocksDBStateBackend类型；另一种基于堆将工作状态保存在Java的堆内存中。基于堆的状态后端有两种类型：FsStateBackend和MemoryStateBackend。FsStateBackend会定时将其状态快照持久化到分布式文件系统中，MemoryStateBackend使用JobManager的堆内存保存状态快照。

Flink执行Checkpoint的流程：

在这里插入图片描述

Flink执行Checkpoint的架构：
在这里插入图片描述
Flink状态后端的分类对比：

5.1.1 RocksDBStateBackend
RocksDBStateBackend状态后端将工作状态(State)存储在RocksDB（一种KV型数据库）中。这个状态后端可以存储超过内存并溢出到磁盘的非常大的状态。所有的key/value状态存储在RocksDB的key/value索引中。为了防止丢失状态数据，Flink将获取RocksDB数据库的快照作为Checkpoint，并在文件系统（默认情况下）或其他可配置状态后端中持久化该快照。可以在Flink应用程序中使用RocksDBStateBackend类中的setPredefinedOptions(PredefinedOptions)方法和setRocksDBOptions(RocksDBOptionsFactory)设置RocksDB的相关选项。

RocksDBStateBackend是目前唯一支持增量Checkpoint（增量快照）的状态后端。不同于产生一个包含所有数据的全量备份，增量快照中只包含自上一次快照完成之后被修改的记录，因此可以显著减少快照完成所耗的时间。

虽然RocksDBStateBackend支持增量快照，但是默认情况下没有开启该功能，使用的仍然是全量快照，如果需
要开启，可以通过在flink-conf.yaml配置文件中设置state.backend.incremental:true实现。

对于超大状态的聚合，例如以天为单位的窗口计算并且对读写性能要求不高的作业，建议使用RocksDBStateBackend。

5.1.2 FsStateBackend
FsStateBackend状态后端在TaskManager的内存（JVM堆）中保存运行时的工作状态。执行Checkpoint时，会将状态数据以文件的形式持久化到外部文件系统中。如果外部文件系统是持久的分布式文件系统，则此状态后端支持高可用设置。每个Checkpoint将分别将其所有文件存储在包含Checkpoint编号的文件系统的子目录中，例如HDFS目录hdfs://namenode:port/flink-checkpoints/chk-17/。

如果一个TaskManager并发执行多个任务（如果TaskManager有多个Task Slot，或者使用Task Slot共享），那么所有任务的聚合状态需要被放入该TaskManager的内存。FsStateBackend状态后端直接与元数据一起存储小的状态块，以避免创建许多小文件。其阈值是可配置的。当增加这个阈值时，Checkpoint元数据的大小也会增加。所有保留的已完成Checkpoint的元数据需要装入JobManager的堆内存。这都不是问题，除非阈值太大。可以通过调用getMinFileSizeThreshold()方法获取设置的阈值。

FsStateBackend状态后端适用于状态比较大、窗口比较长的作业以及所有高可用的场景。对于一些以分钟为单位的窗口聚合，建议使用该状态后端。

5.1.3 MemoryStateBackend

MemoryStateBackend状态后端在TaskManager的内存（JVM堆）中以Java对象的形式保存运行时的工作状态。执行Checkpoint时，会直接将其状态保存到JobManager的堆内存。默认每个状态在JobManager中允许使用的最大内存为5MB，可以通过MemoryStateBackend的构造函数进行调整。

该状态后端建议只用于实验、本地测试或状态数据非常小的流应用程序，因为它需要将Checkpoint数据存储在JobManager的内存中，较大的状态数据将占用较大一部分JobManager的主内存，从而降低操作的稳定性。对于任何其他设置，都应该使用FsStateBackend。FsStateBackend将工作状态以同样的方式保存在TaskManager上，但执行Checkpoint时状态数据直接存储在文件系统中，而不是JobManager的内存中，因此支持非常大的状态数据。

所有状态后端都可以在应用程序中配置（通过使用各自的构造函数参数创建状态后端并在执行环境中设置），也可以在Flink集群环境中指定。如果在应用程序中指定了状态后端，则它可以从Flink集群环境配置中获取额外的配置参数。例如，如果在没有默认保存点（Savepoint，在4.13.4节将详细讲解）目录的应用程序中，则它将选择在运行的集群环境的Flink配置中指定的默认保存点目录。

通常，建议在生产中避免使用MemoryStateBackend，因为它将快照存储在JobManager的内存中，而不是持久化到磁盘。当需要在FsStateBackend和RocksDBStateBackend之间进行选择时，需要从性能和可伸缩性方面进行考虑。FsStateBackend非常快，因为每个状态访问和更新操作在Java堆内存上，但是状态大小受集群内可用内存的限制。另一方面，RocksDBStateBackend可以根据可用磁盘空间进行扩展，并且是唯一支持增量快照的状态后端。但是每个状态访问和更新都需要序列化／反序列化，这样会导致平均性能比内存状态后端慢一个数量级。

5.2 状态后端配置

可以在配置文件flink-conf.yaml中通过属性state.backend设置全局默认的状态后端。例如，设置状态后端为FsStateBackend：

#使用文件系统存储快照
state.backend: filesystem
#存储快照的目录
state.checkpoints.dir: hdfs://namenode:40010/flink/checkpoints

state.backend的可选值包括jobmanager（MemoryStateBackend状态后端）、filesystem（FsStateBackend状态后端）、rocksdb（RocksDBStateBackend状态后端），或使用实现了状态后端工厂StateBackendFactory的类的全限定类名，例如RocksDBStateBackend对应org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory。

state.checkpoints.dir指定了所有状态后端的数据存储目录。

也可以在应用程序中使用StreamExecutionEnvironment API对作业的状态后端进行设置。从Flink 1.13开始，在API层面为了对状态后端更容易理解，重新编写了状态后端的公共类，以帮助开发者更好地理解本地状态存储和检查点存储的分离。用户可以在不丢失任何状态或一致性的情况下迁移现有应用程序以使用新的API。

例如，设置状态后端为FsStateBackend，代码如下：
在这里插入图片描述
设置状态后端为RocksDBStateBackend，代码如下：

设置状态后端为MemoryStateBackend，代码如下：

val env=StreamExecutionEnvironment.getExecutionEnvironment
env.setStateBackend(new HashMapStateBackend)
env.getCheckpointConfig.setCheckpointStorage(new JobManagerCheckpointStorage)

上述代码中，HashMapStateBackend在TaskManager的内存（JVM堆）中保存运行时的工作状态。执行Checkpoint时，会根据配置的CheckpointStorage保存状态到指定的位置。CheckpointStorage是一个接口，定义了状态后端如何存储其状态以在流应用程序中进行容错。该接口的各种实现以不同的方式存储检查点状态，并具有不同的可用性保证。

例如，JobManagerCheckpointStorage将检查点数据存储在JobManager的内存中。它是轻量级的，没有额外的依赖项，但不可扩展，只支持小量状态数据。这种检查点存储策略便于本地测试和开发。

FileSystemCheckpointStorage将检查点存储在HDFS、S3等文件系统中。此存储策略支持大量状态数据，可以达到数TB，同时为有状态应用程序提供高度可用的基础。对于大多数生产部署，建议使用此检查点存储策略。

5.3 Checkpoint配置

默认情况下，Checkpoint是禁用的，可以通过相应的配置启用。

5.3.1 全局配置

可以在Flink的配置文件flink-conf.yaml中对Checkpoint进行全局配置，代码如下：

 state.backend: filesystem
 state.checkpoints.dir: hdfs://namenode:9000/flink-checkpoints
 state.backend.incremental: false

state.backend用于指定状态后端，支持的值为jobmanager、filesystem、rocksdb，常用值为filesystem或rocksdb，默认为none。jobmanager表示使用MemoryStateBackend状态后端，filesystem表示使用FsStateBackend状态后端，rocksdb表示使用RocksDBStateBackend状态后端。

state.checkpoints.dir用于指定Checkpoint在文件系统的存储目录，默认为none。

state.backend.incremental用于开启／禁用增量Checkpoint功能，默认为false。对于支持增量Checkpoint的状态后端有用，例如RocksDBStateBackend。

常见Flink Checkpoint全局配置选项介绍：

在这里插入图片描述

5.3.2 应用配置

除了可以在Flink的配置文件中对Checkpoint进行全局配置外，还可以在Flink应用程序中通过代码配置Checkpoint，该配置将覆盖配置文件中的全局配置。

在Flink应用程序中进行Checkpoint配置，必须的配置选项如下：

val env=StreamExecutionEnvironment.getExecutionEnvironment
//每隔1秒执行一次Checkpoint
env.enableCheckpointing(1000)
//指定状态后端
env.setStateBackend(new FsStateBackend("file:///D:checkpoint"))

其他可选选项说明如下：

当设置了每隔一秒执行一次Checkpoint时，如果由于网络延迟，前一次Checkpoint执行较慢，容易导致与后一次Checkpoint执行重叠。为了防止这种情况，可以设置两次Checkpoint之间的最小时间间隔：

//设置两次Checkpoint之间的最小时间间隔为500毫秒，默认为0
env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)

设置可容忍的失败的Checkpoint数量，默认值为0，意味着不容忍任何Checkpoint失败：

env.getCheckpointConfig.setTolerableCheckpointFailureNumber(10)

作业取消时保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint：

env.getCheckpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)

ExternalizedCheckpointCleanup的相关选项如下：

ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION：在作业取消时保留Checkpoint数据。取消作业时，保留所有的Checkpoint数据。在取消作业之后，必须手动删除Checkpoint数据。
ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION：在作业取消时删除Checkpoint数据（包括元数据和实际的状态数据），删除后不能进行恢复。作业失败时不会删除。

CheckpointingMode定义了系统在出现故障时提供的一致性保证。例如，设置Checkpoint执行模式为exactly once（默认值）：

env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

设置Checkpoint执行模式为at least once：

env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);

设置Checkpoint执行的超时时间为一分钟，超过该时间则被丢弃，默认超时时间为10分钟：

env.getCheckpointConfig.setCheckpointTimeout(6000)

设置最大允许的同时执行的Checkpoint的数量，默认为1。当达到设置的最大值时，如果需要触发新的Checkpoint，需要等待正在执行的Checkpoint完成或过期：

env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);