Flink状态管理和恢复机制

Posted 2023-03-09

参考技术A

1、什么是状态？
2、Flink状态类型有哪几种？
3、状态有什么作用？
4、如何使用状态，实现什么样的API？
5、什么是checkpoint与savepoint？
6、如何使用checkpoint与savepoint？
7、checkpoint原理是什么？
8、checkpint存储到hdfs上又是什么意思？

<1> 增量计算
聚合操作、机器学习训练模型迭代运算时保存当前模型等等
<2> 容错
Job故障重启、升级

定义： 某task或者operator 在某一时刻的在内存中的状态。
而checkpoint是，对于这个中间结果进行一次快照。
作用：State是可以被记录的，在失败的情况下可以恢复。
checkpoint则表示了一个Flink Job，在一个特定时刻的一份全局状态快照，即包含了一个job下所有task/operator某时刻的状态。

比如任务挂掉的时候或被手动停止的时候，可以从挂掉的点重新继续消费。
基本类型：Operator state、Keyed state
特殊的 Broadcast State
适用场景：
增量计算：
<1>聚合操作
<2>机器学习训练模型迭代运算时保存当前模型
等等
容错：
Job故障重启

使用状态，必须使用RichFunction，因为状态是使用RuntimeContext访问的，只能在RichFunction中访问

假设现在存在输入源数据格式为(EventID,Value)
输出数据，直接flatMap即可，无状态。
如果要输出某EventID最大值/最小值等,HashMap是否可以？
程序一旦Crash，如何恢复？

答案：Flink提供了一套状态保存的方法，不需要借助第三方存储系统来解决状态存储问题。

Operator State跟一个特定operator的一个并发实例绑定，整个operator只对应一个state。相比较而言，在一个operator上，可能有很多个key，从而对应多个keyed state。
所以一个并行度为4的source，即有4个实例，那么就会有4个状态

举例：Flink中的Kafka Connector,就使用了operator state。有几个并行度，就会有几个connector实例，消费的分区不一样，它会在每个connector实例中，保存该实例中消费topic的所有(partition，offset)映射。

数据结构：ListState<T>

一般编码过程：实现CheckpointedFunction接口,必须实现两个函数，分别是：
initializeState和snapshotState

如何保存状态？
通常是定义一个private transient ListState<Long> checkPointList;

注意：使用Operator State最好不要在keyBy之后使用，另外不要将太大的state存放到这个里面。

是基于KeyStream之上的状态，keyBy之后的Operator State。
那么，一个并行度为3的keyed Opreator有几个状态，这个就不一定是3了，这里有几个状态是由keyby之后有几个key所决定的。

案例：有一个事件流Tuple2[eventId,val],求不同的事件eventId下，相邻3个val的平均值，事件流如下：
(1,4),(2,3),(3,1),(1,2),(3,2),(1,2),(2,2),(2,9)
那么事件1:8/3=2
那么事件2:14/3=4

Keyed State的数据结构类型有:
ValueState<T>:update(T)
ListState<T>:add(T)、get(T)和clear(T)
ReducingState<T>:add(T)、reduceFunction()
MapState<UK,UV>:put(UK,UV)、putAll(Map<UK,UV>)、get(UK)

FlatMapFunction是无状态函数；RichFlatMapFunction是有状态函数

这里没有实现CheckpointedFunction接口，而是直接调用方法 getRuntimeContext(),然后使用getState方法来获取状态值。

特殊场景： 来自一个流的一些数据需要广播到所有下游任务，在这些任务中，这些数据被本地存储并且用于处理另一个流上的所有处理元素 。例如：一个低吞吐量流，其中包含一组规则，我们希望对来自另一个流的所有元素按照规则进行计算

典型应用：常规事件流.connect(规则流)
常规事件流.connect(配置流)

<1> 创建常规事件流DataStream或者KeyedDataStream
<2> 创建BroadcastedStream：创建规则流/配置流(低吞吐)并广播
<3> 连接两个Stream并实现计算处理
process(可以是BroadcastProcessFunction 或者 KeyedBroadcastProcessFunction )

BroadcastProcessFunction：

processElement(...)：负责处理非广播流中的传入元素

processBroadcastElement(...)：负责处理广播流中的传入元素(如规则),一般广播流的元素添加到状态里去备用，processElement处理业务数据时就可以使用

ReadOnlyContext和Context：
ReadOnlyContext对Broadcast State只有只读权限，Conetxt有写权限

KeyedBroadcastProcessFunction：

注意：
<1> Flink之间没有跨Task的通信
<2> 每个任务的广播状态的元素顺序有可能不一样
<3> Broadcast State保存在内存中(并不在RocksDB)