6 Flink 状态管理

我们前面写的 wordcount 的例子，没有包含状态管理。如果一个task在处理过程中挂掉了，那么它在内存中的状态都会丢失，所有的数据都需要重新计算。从容错和消息处理的语义上(at least once, exactly once)，Flink引入了state和checkpoint。
因此可以说flink因为引入了state和checkpoint所以才支持的exactly once
首先区分一下两个概念：

state：
state一般指一个具体的task/operator的状态：
state数据默认保存在java的堆内存中，TaskManage节点的内存中。
operator表示一些算子在运行的过程中会产生的一些中间结果。

checkpoint：
checkpoint可以理解为checkpoint是把state数据定时持久化存储了，则表示了一个Flink Job在一个特定时刻的一份全局状态快照，即包含了所有task/operator的状态。
注意：task(subTask)是Flink中执行的基本单位。operator指算子(transformation)
State可以被记录，在失败的情况下数据还可以恢复。

Flink中有两种基本类型的State：
Keyed State
Operator State

Keyed State和Operator State，可以以两种形式存在：
原始状态(raw state)
托管状态(managed state)
托管状态是由Flink框架管理的状态。

我们说operator算子保存了数据的中间结果，中间结果保存在什么类型中，如果我们这里是托管状态，则由flink框架自行管理
原始状态由用户自行管理状态具体的数据结构，框架在做checkpoint的时候，使用byte[]来读写状态内容，对其内部数据结构一无所知。
通常在DataStream上的状态推荐使用托管的状态，当实现一个用户自定义的operator时，会使用到原始状态。

1. State-Keyed State

基于KeyedStream上的状态。这个状态是跟特定的key绑定的，对KeyedStream流上的每一个key，都对应一个state，比如：stream.keyBy(…)。KeyBy之后的Operator State,可以理解为分区过的Operator State。
保存state的数据结构：
ValueState：即类型为T的单值状态。这个状态与对应的key绑定，是最简单的状态了。它可以通过update方法更新状态值，通过value()方法获取状态值。
ListState：即key上的状态值为一个列表。可以通过add方法往列表中附加值；也可以通过get()方法返回一个Iterable来遍历状态值。
ReducingState:这种状态通过用户传入的reduceFunction，每次调用add方法添加值的时候，会调用reduceFunction，最后合并到一个单一的状态值。
MapState<UK, UV>:即状态值为一个map。用户通过put或putAll方法添加元素。
需要注意的是，以上所述的State对象，仅仅用于与状态进行交互（更新、删除、清空等），而真正的状态值，有可能是存在内存、磁盘、或者其他分布式存储系统中。相当于我们只是持有了这个状态的句柄。

1. ValueState
   使用ValueState保存中间结果对下面数据进行分组求和。
   开发步骤：
2. 获取流处理执行环境
3. 加载数据源
4. 数据分组
5. 数据转换，定义ValueState,保存中间结果
6. 数据打印
7. 触发执行
   ValueState:测试数据源：

 List((1L, 4L),(2L, 3L),(3L, 1L),(1L, 2L),(3L, 2L),(1L, 2L),(2L, 2L),(2L, 9L)
)

示例代码：

import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.common.typeinfo.{TypeHint, TypeInformation}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.util.Collectorobject TestKeyedState {class CountWithKeyedState extends RichFlatMapFunction[(Long, Long), (Long, Long)] {/*** ValueState状态句柄. 第一个值为count，第二个值为sum。*/private var sum: ValueState[(Long, Long)] = _override def flatMap(input: (Long, Long), out: Collector[(Long, Long)]): Unit = {// 获取当前状态值val tmpCurrentSum: (Long, Long) = sum.value// 状态默认值val currentSum = if (tmpCurrentSum != null) {tmpCurrentSum} else {(0L, 0L)}// 更新val newSum = (currentSum._1 + 1, currentSum._2 + input._2)// 更新状态值sum.update(newSum)// 如果count >=3 清空状态值，重新计算if (newSum._1 >= 3) {out.collect((input._1, newSum._2 / newSum._1))sum.clear()}}override def open(parameters: Configuration): Unit = {sum = getRuntimeContext.getState(new ValueStateDescriptor[(Long, Long)]("average", // 状态名称TypeInformation.of(new TypeHint[(Long, Long)](){}) )// 状态类型)}}  def main(args: Array[String]): Unit = {//初始化执行环境val env = StreamExecutionEnvironment.getExecutionEnvironment//构建数据源val inputStream: DataStream[(Long, Long)] = env.fromCollection(List((1L, 4L),(2L, 3L),(3L, 1L),(1L, 2L),(3L, 2L),(1L, 2L),(2L, 2L),(2L, 9L)))//执行数据处理inputStream.keyBy(0).flatMap(new CountWithKeyedState).setParallelism(1).print//运行任务env.execute}
}  

2. MapState
   使用MapState保存中间结果对下面数据进行分组求和:
3. 获取流处理执行环境
4. 加载数据源
5. 数据分组
6. 数据转换，定义MapState,保存中间结果
7. 数据打印
8. 触发执行
   MapState:测试数据源：

List(("java", 1),("python", 3),("java", 2),("scala", 2),("python", 1),("java", 1),("scala", 2)
)  

示例代码：

object MapState {def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironmentenv.setParallelism(1)/*** 使用MapState保存中间结果对下面数据进行分组求和* 1.获取流处理执行环境* 2.加载数据源* 3.数据分组* 4.数据转换，定义MapState,保存中间结果* 5.数据打印* 6.触发执行*/val source: DataStream[(String, Int)] = env.fromCollection(List(("java", 1),("python", 3),("java", 2),("scala", 2),("python", 1),("java", 1),("scala", 2)))source.keyBy(0).map(new RichMapFunction[(String, Int), (String, Int)] {var mste: MapState[String, Int] = _override def open(parameters: Configuration): Unit = {val msState = new MapStateDescriptor[String, Int]("ms",TypeInformation.of(new TypeHint[(String)] {}),TypeInformation.of(new TypeHint[(Int)] {}))mste = getRuntimeContext.getMapState(msState)}override def map(value: (String, Int)): (String, Int) = {val i: Int = mste.get(value._1)mste.put(value._1, value._2 + i)(value._1, value._2 + i)}}).print()env.execute()}
}  

2. State-Operator State

与Key无关的State，与Operator绑定的state，整个operator只对应一个state。
保存state的数据结构：
ListState
举例来说，Flink中的 Kafka Connector，就使用了operator state。它会在每个connector实例中，保存该实例中消费topic的所有(partition, offset)映射。
步骤：
1.获取执行环境
2.设置检查点机制：路径，重启策略
3.自定义数据源
需要继承并行数据源和CheckpointedFunction
设置listState,通过上下文对象context获取
数据处理，保留offset
制作快照
4.数据打印
5.触发执行
示例代码：

import java.utilimport org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.api.common.time.Time
import org.apache.flink.api.common.typeinfo.{TypeHint, TypeInformation}
import org.apache.flink.runtime.state.{FunctionInitializationContext, FunctionSnapshotContext}
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction
import org.apache.flink.streaming.api.environment.CheckpointConfig
import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._object ListOperate {def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironmentenv.setParallelism(1)env.enableCheckpointing(5000)env.setStateBackend(new FsStateBackend("hdfs://node01:8020/tmp/check/8"))env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)env.getCheckpointConfig.setCheckpointTimeout(60000)env.getCheckpointConfig.setFailOnCheckpointingErrors(false)env.getCheckpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)//重启策略env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.minutes(1), Time.seconds(5)))//模拟kakfa偏移量env.addSource(new MyRichParrelSourceFun).print()env.execute()}}class MyRichParrelSourceFun extends RichParallelSourceFunction[String]with CheckpointedFunction {var listState: ListState[Long] = _var offset: Long = 0L//任务运行override def run(ctx: SourceFunction.SourceContext[String]): Unit = {val iterState: util.Iterator[Long] = listState.get().iterator()while (iterState.hasNext) {offset = iterState.next()}while (true) {offset += 1ctx.collect("offset:"+offset)Thread.sleep(1000)if(offset > 10){1/0}}}//取消任务override def cancel(): Unit = ???//制作快照override def snapshotState(context: FunctionSnapshotContext): Unit = {listState.clear()listState.add(offset)}//初始化状态override def initializeState(context: FunctionInitializationContext): Unit = {listState = context.getOperatorStateStore.getListState(new ListStateDescriptor[Long]("listState", TypeInformation.of(new TypeHint[Long] {})))}
}

3. Broadcast State

Broadcast State 是 Flink 1.5 引入的新特性。在开发过程中，如果遇到需要下发/广播配置、规则等低吞吐事件流到下游所有 task 时，就可以使用 Broadcast State 特性。下游的 task 接收这些配置、规则并保存为 BroadcastState, 将这些配置应用到另一个数据流的计算中 。

1. API介绍
   通常，我们首先会创建一个Keyed或Non-Keyed的Data Stream，然后再创建一个Broadcasted Stream，最后通过Data Stream来连接（调用connect方法）到Broadcasted Stream上，这样实现将Broadcast State广播到Data Stream下游的每个Task中。
   如果Data Stream是Keyed Stream，则连接到Broadcasted Stream后，添加处理ProcessFunction时需要使用KeyedBroadcastProcessFunction来实现，下面是KeyedBroadcastProcessFunction的API，代码如下所示：

public abstract class KeyedBroadcastProcessFunction<KS, IN1, IN2, OUT> extends BaseBroadcastProcessFunction {public abstract void processElement(final IN1 value, final ReadOnlyContext ctx, final Collector<OUT> out) throws Exception;public abstract void processBroadcastElement(final IN2 value, final Context ctx, final Collector<OUT> out) throws Exception;
}

上面泛型中的各个参数的含义，说明如下：
KS：表示Flink程序从最上游的Source Operator开始构建Stream，当调用keyBy时所依赖的Key的类型；
IN1：表示非Broadcast的Data Stream中的数据记录的类型；
IN2：表示Broadcast Stream中的数据记录的类型；
OUT：表示经过KeyedBroadcastProcessFunction的processElement()和processBroadcastElement()方法处理后输出结果数据记录的类型。
如果Data Stream是Non-Keyed Stream，则连接到Broadcasted Stream后，添加处理ProcessFunction时需要使用BroadcastProcessFunction来实现，下面是BroadcastProcessFunction的API，代码如下所示：

public abstract class BroadcastProcessFunction<IN1, IN2, OUT> extends BaseBroadcastProcessFunction {public abstract void processElement(final IN1 value, final ReadOnlyContext ctx, final Collector<OUT> out) throws Exception;public abstract void processBroadcastElement(final IN2 value, final Context ctx, final Collector<OUT> out) throws Exception;}

上面泛型中的各个参数的含义，与前面KeyedBroadcastProcessFunction的泛型类型中的后3个含义相同，只是没有调用keyBy操作对原始Stream进行分区操作，就不需要KS泛型参数。
注意事项：
1.Broadcast State 是Map类型，即K-V类型。
2.Broadcast State 只有在广播一侧的方法中processBroadcastElement可以修改;在非广播一侧方法中processElement只读。
3.Broadcast State在运行时保存在内存中。
2) 场景举例
1.动态更新计算规则: 如事件流需要根据最新的规则进行计算，则可将规则作为广播状态广播到下游Task中。
2.实时增加额外字段: 如事件流需要实时增加用户的基础信息，则可将用户的基础信息作为广播状态广播到下游Task中。