Flink实战五_状态机制

接上文：Flink实战四_TableAPI&SQL

在学习Flink的状态机制之前，我们需要理解什么是状态。回顾我们之前介绍的很多流计算的计算过程，有些计算方法，比如说我们之前多次使用的将stock.txt中的一行文本数据转换成Stock股票对象的map操作。来一个数据，就计算一个数据，这些操作，只需要依赖于当前输入的数据就够了，不需要其他的辅助数据。输入相同的文本数据，输出的肯定是一个相同的Stock股票对象。

而另外一些操作，比如在WindowFunctionDemo中介绍的计算平均值的方法。同样是来一个数据处理一个数据，但是，他每次计算出来的结果，除了依赖于当前输入的数据，还需要依赖于accumulate累加器中的数据。输入相同的股票数据，由于累加器中的数据不同，输出的股票平均价格也就不同。

像累加器这种由一个任务来维护，并且要参与到数据计算过程中的数据，就称为状态。这一类计算任务，也称为有状态的任务。比如reduce、sum、min、minby等等操作，都是典型的有状态的算子。而与之对应的，只依赖于输入数据的计算任务，就称为无状态的任务。多个任务叠加在一起，就组成了一个客户端应用。

对于状态，也可以认为就是一个本地变量，他可以被一个客户端应用中的所有计算任务都访问到。对于状态的管理通常是比较复杂的，尤其在分布式流式计算场景下。任务是并行计算的，所以状态也需要分开保存。集群故障恢复后又需要合并读取。在算子并行度发生变化时又要维护状态的一致性。再考虑到状态数据要尽量高效的存储与访问，等等。Flink的状态机制提供了对这类状态数据的统一管理。开发人员可以专注于开发业务逻辑，而不用时刻考虑状态的各种复杂管理机制。

对于状态，有两种管理机制，一种是managed state，就是Flink管理的状态机制，对之前提到的一些状态管理的问题提供了统一的管理机制。另一种是raw state，就是用户自己管理的状态机制。只需要Flink提供一个本地变量空间，由应用程序自己去管理这一部分状态。Flink的状态管理机制非常强大，所以在大部分的开发场景下，我们使用Flink提供的状态管理机制就足够了。

Flink中管理的状态都是跟特定计算任务关联在一起的。他的状态主要有两种，一种是operator state 算子状态，一种是keyed State 键控状态。

1、Operator State 算子状态

算子状态的作用范围限定为当前计算任务内，这种状态是跟一个特定的计算任务绑定的。算子状态的作用范围只限定在算子任务内，由同一并行任务所处理的所有数据都可以访问到相同的状态。并且这个算子状态不能由其他子任务访问。比如WindowFunctionDemo中计算股票平均价格的MyAvg计算任务里的累加器，就只能在当前计算任务中访问。即使在多个不同的应用程序中都可以使用MyAvg这个计算任务，但是每个应用程序中访问到的累加器都是不同的。

这一类算子需要按任务分开保存，而当任务的并行度发生变化时，还需要支持在并行运算实例之间，重新分配状态。

例如下面我们定义一个带状态的求和算子，在这个示例中就给一个简单的求和算子保存了一个状态。

示例代码： com.flink.state.SumOperatorState

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 算子状态
 */
public class SumOperatorState {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        env.setStateBackend(new RocksDBStateBackend("hdfs://hadoop01:8020/SumOperatorState"));
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.of(10, TimeUnit.SECONDS)));
        final DataStreamSource<Integer> stream = env.fromElements(1, 2, 3, 4, 5,6);
        final SingleOutputStreamOperator<Integer> stream2 = stream.map(new MySumMapper("mysummapper"));
        stream2.print();
//        final DataStream<Integer> union = stream.union(stream2);
        env.execute("stream");
    }

    public static class MySumMapper implements MapFunction<Integer,Integer>, CheckpointedFunction {

        private int sum;
        private String stateKey;
        private ListState<Integer> checkpointedState;

        public MySumMapper(String stateKey){
            this.stateKey = stateKey;
        }

        @Override
        public Integer map(Integer value) throws Exception {
            return sum += value;
        }

        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            checkpointedState.clear();
            checkpointedState.add(sum);
        }

        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            ListStateDescriptor<Integer> descriptor = new ListStateDescriptor<Integer>(
                    stateKey,
                    TypeInformation.of(new TypeHint<Integer>() {
            }));
            checkpointedState = context.getOperatorStateStore().getListState(descriptor);
            if(context.isRestored()){
                for (Integer subSum : checkpointedState.get()) {
                    sum += subSum;
                }
            }
        }
    }
}

可以看到，Flink中的算子状态操作还是比较简单的，可以给算子继承一个CheckpointedFunction接口，这个接口有两个方法，snapshotState方法会在算子执行过程中调用，进行状态保存。initializeState方法是在任务启动时加载初始化的状态。这样，算子在执行过程中，就可以将中间结果保存到checkpointedState状态中。当算子异常终止时，下一次启动又可以从这个checkpointedState状态中加载之前的计算结果。用户程序只需要定义逻辑，而不需要管触发的时机。

关于不同的状态类型：
在获取状态的地方： context.getOperatorStateStore()这个方法有几个重载的方法：getListState，getUnionListState，getBroadcastState。

其中，getListState和getUnionListState，这两个方法都是处理ListState，也就是不同的任务节点，他的状态也不相同。只是这两种状态的底层状态分配机制不同。ListState是将不同的子状态分配好了之后，分给不同的算子实例去处理。而
UnionListState则是将所有的子状态都分配给所有的算子实例，由算子实例自行调节每个实例获取哪些状态。FlinkKafkaConsumer就是使用的UnionListState。

最后一个getBoradcastState，是处理广播状态，也就是所有任务节点的状态都是一样的。

其他的算子，包括function,source,sink都可以自行添加状态管理。这其中需要理解的就是checkpointedState的形式，为什么是一个集合状态ListState？

在这里插入图片描述
这是因为Flink的计算任务都是并行执行的，那么在计算过程中，每一个并行的实例都会有一个自己的状态，所以在snapshotState保存状态时，是将每个并行实例内的状态进行保存，那整个任务整体就会保存成一个集合。所以，示例中保存的其实是每个子任务内计算到的sum和。

当任务重新启动时，Flink可能还需要对子任务的状态进行重新分配，因为任务的并行度有可能进行了调整。所以示例中initializeState方法加载状态时，也是将各个子状态的sum加到一起，才是一个完整的求和计算。

2、keyed State 键控状态

算子状态针对的是普通算子，在任何DataStream和DataSet中都可以使用。但是，如果针对KeyedStream，情况又有所不同。相比算子状态，keyedState键控状态是针对keyby产生的KeyedStream。KeyedStream的计算任务都跟当前分配的key直接关联。相对应的KeyedState状态也就跟key有关。而key是在计算任务运行时分配的。这一类状态，无法在任务启动过程中完成状态的分配。需要在任务执行过程中，根据key的分配不同而进行不同的分配。Flink针对keyedStream，会在内部根据每个key维护一个键控状态。在具体运算过程中，根据key的分配情况，将状态分配给不同的计算任务。

针对键控状态， Flink提供了一系列Rich开头的富计算因子抽象类，这些抽象类提供了更丰富的计算任务生命周期管理。用户程序通过继承这些抽象类，就可以获取到与当前分配的key相关的状态。

我们先来看一个关于KeyedStream的状态示例。下面实现了一个自定义的求word count的算子。

示例代码： com.flink.state.WCKeyedState

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author roy
 * @date 2021/9/6
 */
public class WCKeyedState {
    public static void main(String[] args) throws Exception {
        //创建执行环境
        final StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        final DataStreamSource<Tuple2<String, Integer>> stream = environment.fromCollection(Arrays.asList(Tuple2.of("a", 1), Tuple2.of("a", 5), Tuple2.of("a", 7), Tuple2.of("a", 2)
                , Tuple2.of("b", 2), Tuple2.of("b", 6), Tuple2.of("b", 3), Tuple2.of("b", 8)
                , Tuple2.of("c", 4), Tuple2.of("c", 8), Tuple2.of("c", 4), Tuple2.of("c", 6)));
        //按照字符分组
        final KeyedStream<Tuple2<String, Integer>, String> keyedStream = stream.keyBy((key) -> key.f0);
        keyedStream.flatMap(new WCFlatMapFunction("WCKeyedState")).print();

        environment.execute("WCKeyedState");
    }

    public static class WCFlatMapFunction extends RichFlatMapFunction<Tuple2<String,Integer>, Tuple2<String,Integer>>{

        private String stateDesc;
        ValueState<Tuple2<String, Integer>> valueState;

        public WCFlatMapFunction(String stateDesc) {
            this.stateDesc = stateDesc;
        }

        @Override
        public void flatMap(Tuple2<String, Integer> input, Collector<Tuple2<String, Integer>> out) throws Exception {
            Tuple2<String, Integer> wordCountList = valueState.value();
            if(null == wordCountList){
                wordCountList = input;
            }else{
                wordCountList.f1+= input.f1;
            }
            valueState.update(wordCountList);
            out.collect(wordCountList);
//            valueState.clear();
        }

        @Override
        public void open(Configuration parameters) {
            ValueStateDescriptor<Tuple2<String,Integer>> descriptor =
                new ValueStateDescriptor<>(stateDesc,
                        TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {}));
            //设置状态的存活时间
            StateTtlConfig ttlConfig = StateTtlConfig
                    .newBuilder(Time.seconds(1))
                    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                    .build();
            descriptor.enableTimeToLive(ttlConfig);
            valueState = this.getRuntimeContext().getState(descriptor);
            //另外几种state类型
//            this.getRuntimeContext().getMapState();
//            this.getRuntimeContext().getListState();
//            this.getRuntimeContext().getAggregatingState();
//            this.getRuntimeContext().getReducingState();
        }

    }
}

在这个示例中看到，其实键控状态与算子状态在应用代码层面最大的区别在于获取状态的方法。算子状态可以通过FunctionInitializationContext直接拿到状态，而键控状态需要实现Rich***Function接口，在open方法中通过getRuntimeContext()获取。

而更深层次的区别在于运行机制上。其实你可以把这个状态近似的理解为一个
(key,value)结构的本地缓存。算子状态的缓存是一个固定的key，只是这个key跟当前计算任务有关，只有当前这一个算子能够读到，不管在哪个taskmanager上计算，如果能读到这个缓存的话，那读到的缓存就是一个固定的。而键控状态的缓存是一组(key,value)的缓存，这一组缓存的key就是KeyedStream中的key分区键。而键控状态获取到的状态值都是取决于当前输入元素所代表的key分区键的，因此，每次任务时taskmanager上分配的key不同，那就可能读取到不同的值。

另外，根据状态类型不同， Flink也提供了几种不同的状态：

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

这些不同的状态都是跟Key相关的。使用时，都需要通过构建一个对应的
StateDescriptor，然后通过getRuntimeContext获取。

3、Checkpointing 检查点

Flink中的每个算子都可以是有状态的，这些状态化的方法和算子可以使Flink的计算过程更为精确，在实际开发中，应该尽量使用带状态的算子。而对于这些状态，除了可以通过算子状态和键控状态进行扩展外，Flink也提供了另外一种自动的兜底机制，CheckPointing检查点。

Checkpointing检查点是一种由Flink自动执行的一种状态备份机制，其目的是能够从故障中恢复。快照中包含了每个数据源Source的指针(例如，到文件或者kafka分区的偏移量)以及每个有状态算子的状态副本。

默认情况下，检查点机制是禁用的，需要在应用中通过StreamExecutionEnvironment 进行配置。基础的配置方式是通过StreamExecutionEnvironment的enableCheckpointing方法开启，开启时需要传入一个参数，表示多长时间执行一次快照。另外有一些高级的选项，可以参见下面的示例。

StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
// 每 1000ms 开始一次 checkpoint
env.enableCheckpointing(1000);

// 高级选项：
// 设置模式为精确一次 (这是默认值)
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ON
CE);
// 确认 checkpoints 之间的时间会进行 500 ms
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
// Checkpoint 必须在一分钟内完成，否则就会被抛弃
env.getCheckpointConfig().setCheckpointTimeout(60000);
// 同一时间只允许一个 checkpoint 进行
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
// 开启在 job 中止后仍然保留的 externalized checkpoints
env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpo
intCleanup.RETAIN_ON_CANCELLATION);

4、Flink的容错重启机制

当某一个task发生故障时，Flink需要重启出错的task以及其他收到影响的Task，以使得作业恢复到正常执行的状态。Flink通过重启策略和故障恢复策略来控制Task重启：重启策略决定是否可以重启以及重启的间隔；故障恢复策略决定哪些Task需要重启。

重启策略可以通过配置文件flink-conf.yaml中通过restart-strategy属性进行配置，同样，也可以在应用程序中覆盖配置文件中的配置。如果没有启用checkpoint，那就采用"不重启"的策略。如果启用了checkpoint并且没有配置重启策略，那么就采用固定延时重启策略，这种情况下最大尝试重启次数是Integer.MAX_VALUE，基本就可以认为是会不停的尝试重启。

restart-strategy属性可选的配置有以下几种：

none 或 off 或 disable: 不重启。checkpointing关闭后的默认值
fixeddelay, fixed-delay: 固定延迟重启策略。checkpointing启用时的默认值
failurerate, failure-rate: 失败率重启策略

这些配置项同样可以在应用程序中定制。例如

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
3, // 尝试重启的次数
Time.of(10, TimeUnit.SECONDS) // 延时
));

fixeddelay策略，还可以定制两个参数，restart-strategy.fixed-delay.attempts 重试次数以及 restart-strategy.fixed-delay.delay延迟时间。第一个参数表示重启任务的尝试次数，第二个参数表示重启失败后，再次尝试重启的间隔时间。可以配置为 “1 min”,"20 s"这样。例如在配置文件中

restart-strategy.fixed-delay.attempts: 3
restart-strategy.fixed-delay.delay: 10 s

或者在应用程序中

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
	3, // 尝试重启的次数
	Time.of(10, TimeUnit.SECONDS) // 延时
));

Failure Rate 策略表示当故障率(每个时间假根发生故障的次数)超过设定的限制时，作业就会最终失败。在连续的两次重启尝试之间，重启策略会等待一段固定长度的时间。

这种策略下，可以定义三个详细的参数。

restart-strategy.failure-rate.max-failures-per-interval：任务失败之前，在固定时间间隔内的最大重启尝试次数。
restart-strategy.failure-rate.failure-rate-interval：检测失败率的窗口间隔。
restart-strategy.failure-rate.delay 两次重启尝试之间的间隔时间。

例如在配置文件中

restart-strategy: failure-rate
restart-strategy.failure-rate.max-failures-per-interval: 3
restart-strategy.failure-rate.failure-rate-interval: 5 min
restart-strategy.failure-rate.delay: 10 s

或者在应用中

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.failureRateRestart(
	3, // 每个时间间隔的最大故障次数
	Time.of(5, TimeUnit.MINUTES), // 测量故障率的时间间隔
	Time.of(10, TimeUnit.SECONDS) // 延时
));

5、State Backend 状态存储方式与位置

通过算子状态，键控状态以及检查点，我们可以对计算过程中的中间状态进行保存。这些保存下来的状态即可以在计算中使用，也可以在计算程序异常终止后恢复计算状态时使用。但是，到目前为止，我们都是直接拿来用，而并没有去关注这些状态数据是以何种方式保存并且是保存在什么地方的。

针对这些状态，Flink提供了多种State Backend 状态后端，用来管理状态数据具体的存储方式与位置。Flink默认提供了三种状态后端：jobmanager,filesystem,rocksdb。设置的方式可以在file-conf.yaml中，通过state.backend属性进行配置。也可以在程序中通过StreamExecutionEnvironment配置。例如：

StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(...);

另外，在flink-conf.yaml中，关于state.backend还有一些扩展的属性，这些属性同样即可以在配置文件中配置，也可以在程序中设置。应用程序中的配置优先级更高。

那这三种状态后端要如何取舍呢？

jobmanager:
jobmanager在后台由一个MemoryStateBackend类来实现，从名字看出，是基于内存实现的。状态信息会保存到TaskManager的JVM堆内存中，而检查点信息则会直接保存到JobManager的内存中。这些检查点信息虽然都是基于内存工作，但是也依然会持久化到文件系统当中。

由于检查点保存在Jobmanager中，会加大taskmanager和jobmanager之间的
网络请求，并且也会加大jobmanager的负担，所以这种方式通常只用于实验场
景或者小状态的本地计算场景。

filesystem:
filesystem在后台由一个FsStateBackend类来实现。他依然是基于内存和文件系统进行状态保存。但是检查点信息是由taskmanager进行保存的。保存的文件地址是可以自行配置的。由于taskmanager上执行的任务是动态分配的，所以通常这个保存地址需要配置成所有taskmanager都能访问到的地方，例如hdfs。而taskmanager上由于会有多个并行任务，所以他们的文件存储地址也会用数字进行版本区分，例如hdfs://namenode:port/flink-checkpoints/chk-17/.filesystem的状态访问很快速，适合那些需要大的堆内存的场景。但是fliesystem是受限于内存和GC的，所以他支持的状态数据大小优先。
rocksdb:
rocksdb在后台是由一个 RocksDBStateBackend 类来实现的。RocksDB是一个访问快速的key-value本地缓存，你可以把他理解为一个本地的Redis。但是他能够基于文件系统提供非常高效的访问。所以是一个非常常用的流式计算持久化工具。使用RocketDB后，状态数据就不再受限于内存，转而受限于硬盘RocketDBStateBackend适合支持非常大的状态信息存储。但是RocksDB毕竟是基于文件系统的，所以他的执行速度会比filesystem稍慢，官方提供的经验是大概比filesystem慢10倍，但是这个速度在大多数场景下，也依然够用了。

注：如果在应用中使用rocksdb，需要引入一个依赖

<dependency>
	<groupId>org.apache.flink</groupId>
	<artifactId>flink-statebackend-rocksdb_2.12</artifactId>
	<version>${flink.version}</version>
</dependency>

然后使用StreamExecuteEnvironment设置

env.setStateBackend(new RocksDBStateBackend("key"));

章节总结
在流式计算的场景下，应用程序通常是无法预知数据何时到来的，只能一直运行随时等待数据接入。这时一旦应用程序突然出错终止，就很容易造成数据丢失。所以在流式计算场景下，我们需要对程序的健壮性做更多的考量。Flink提供了一系列的状态机制来加强程序的健壮性。但是在重要的生产环境中，我们对程序健壮性做再多的考量都是不过分的，因此通常还需要加上一些基于运维的监控机制，例如监控flink的进程，监控yarn中的任务状态等，来了进一步保证流式计算程序的安全。