Flink状态专题：keyed state和Operator state

        众所周知，flink是有状态的计算。所以学习flink不可不知状态。
        正好最近公司有个需求，要用到flink的状态计算，需求是这样的，收集数据库新增的数据。
        听起来很简单对吧？起初我也这么认为，现在发现，这尼玛就是变相的动态读取啊。
　　因为数据是一直在增加的，你需要记录这次收集的结果，用于下一次的运算，所以要用到状态计算。
　　废话不多说，直接上干货。
　　关于什么是有状态的flink计算，官方给出的回答是这样的：在flink程序内部存储计算产生的中间结果，并提供给Function或算子计算结果使用。
了解了定义，我们接下来进入主题。 　　

　　1.状态类型

　　　　在Flink中根据数据集是否根据Key进行分区，将状态分为Keyde state和Operator State两种类型。

　　　　（1）Keyed State

　　　　　　表示和key相关的一种state,只能用于KeyedStream类型数据集对应的Functions和Operators之上。Keyed State是Operator State的特例，区别在于Keyed State事先按照

　　　　

　　　　key对数据集进行了分区，每个Key State仅对应一个Operator和Key的组合。Keyed State可以通过Key Groups进行管理，主要用于当算子并行度发生变化时，自动重新分布

　　　　

　　　　Keyed State数据。在系统运行过程种，一个Keyed算子实例可能运行一个或者多个Key Groups 的 keys。

　　　　（2）Operator State

　　　　　　与Keyed State不同的是，Operator State只和并行的算子实例绑定，和数据元素种的key无关，每个算子实例中持有所有数据元素中的一部分状态数据。Operator State   　　　　支持当算子实例并行度发生变化时自动重新分配状态数据。   　　　　在Flink中，Keyed State和Operator State均具有两种形式，托管状态和原生状态。（两种状态有什么不同就不啰嗦了，看的头疼）  

　　2.Managed Keyed State

　　　　Flink中有以下Managed Keyed State类型可以使用。ValueState[T],ListState[T],MapState[K,V]。

　　　　（1）Stateful Function定义

　　　　接下来通过完整的实例来说明如何在RichFlatmapFunction中使用ValueState,完成对介入数据最小值的获取。

　　　　

StreamExecutionEnvironment env = StreamExecutionEnvironment .getExecutionEnvironment;
//创建元素数据集
DataStream<int,long> inputStream = env.fromElements((2,21L),(4,1L),(5,4L));
inputStream.keyBy(“1”).flatMap{
//定义和创建RichFlatMapFunction,第一个参数位输入数据类型，第二个参数位输出数据类型
new RichFlatMapFunction<Map(int,long),Map(int,Map(long,long))>(){
private ValueState leastValueState = null;
@Override
open(Configuration parameters){
ValueStateDescriptor leastValueStateDescriptor =new ValueStateDescriptor ("leastValueState ",class.of(long));
　　 leastValueState = getRuntimeContext.getState(leastValueStateDescriptor );
}

@Override
flatMap(Collector collector,Tuple2(int,long) t){
　　long leastValue =leastValueState .value();
　　if(t.f1>leastValue){
　　　　collector.collect(t,leastValue);
　　}else{
　　　　leastValueState.update(t.f1);
　　　　collector.collect(t,leastValue);
　　}
　}
}
}

　　3.Managed Operator State

　　Operator State是一种non-keyed state，与并行的操作算子实际相关联，例如在Kafka Connector中，每个Kafka消费端算子实例都对应到Kafka的一个分区中，维护Topic分区和

Offsets偏移量作为算子的Operator State。在Flink中可以实现CheckpointedFunction或者ListCheckpointed两个接口来定义操作Managed Operator State的函数。

　　（1）通过CheckpointedFunction接口操作Operator State

        CheckpointedFunction接口定义：

public interface CheckpointedFunction{
//触发checkpoint调用
void snapshotState(FunctionSnapshotContext context)throws Exception;
//每次自定义函数初始化时，调用
void initializeState(FunctionInitializationContext context)throws Exception;
}

　　在每个算子中Managed Operator State都是以List形式存储，算子和算子之间的状态数据相互独立，List存储比较适合状态数据的重新分布，Flink目前支持对Managed Operator

State两种重分布的策略，分别是Even-split Redistribution和Union Redistribution。

　　可以通过实现FlatMapFunction和CheckpointedFunction完成对输入数据中每个key的数据元素数量和算子的元素数量的统计。

　　在initializeState（）方法中分别简历keyedState和operator State两种状态，存储基于Key相关的状态值以及基于算子的状态值。

private class CheckpointCount(int numElements)extends FlatMapFunction<Map(int,long),Map(int,Map(long,long))>with CheckpointedFunction{
//定义算子实例本地变量，存储Operator数据数量
private long operatorCount = null;
//定义keyedState，存储和key相关的状态值
private ValueState keyedState =null;
//定义operatorState,存储算子的状态值
private ListState operatorState = null;
@Override
flatMap(Tuple(int,long)t,Collector collector){
long keyedCount okeyedState.value() +1;
//更新keyedState数量
keyedState.update(keyedCount);
//更新本地算子operatorCount值
operatorCount =operatorCount+1;
//输出结果，包括id,id对应的数量统计keyedCount,算子输入数据的数量统计operatorCount
collector.collect(t.f0,keyedCount,operatorCount);

}
//初始化状态数据
@Override
initializeState(FunctionInitializationContext context){
//定义并获取keyedState
ValueStateDescriptor KeyedDescriptor =new ValueStateDescriptor ("keyedState",createTypeInformation);
keyedState = context.getKeyedStateStore.getState(KeyedDescriptor );
//定义并获取operatorState
ValueStateDescriptor OperatorDescriptor =new ValueStateDescriptor ("OperatorState",createTypeInformation);
operatorState = context.getOperatorStateStore.getListState();
//定义在Restored过程中，从operatorState中回复数据的逻辑
if(context.isRestored){
operatorCount = operatorState.get()
}
//当发生snapshot时，将operatorCount添加到operatorState中
@Override
snapshotState(FunctionSnapshotContext context){
operatorState.clear();
operatorState.add(operatorCount);
}
}
}

可以从上述代码看到，在snapshotState()方法中清理掉上一次checkpoint中存储的operatorState的数据，然后再添加并更新本次算子中需要checkpoint的operatorCount状态变量。当

重启时会调用initializeState方法，重新恢复keyedState和OperatorState,其中operatorCount数据可以从最新的operatorState中恢复。

（2）通过ListCheckpointed接口定义Operator State

  　 ListCheckpointed接口和CheckpointedFunction接口相比再灵活性上相对较弱一点，只能支持List类型的状态，并且在数据恢复时仅支持even-redistribution策略。

　　需要实现以下两个方法来操作Operator State:

　　

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

　　其中snapshotState方法定义数据元素List存储到checkpoints的逻辑，restoreState方法则定义从checkpoints中恢复状态的逻辑。

class numberRecordsCount extends FlatMapFunction(Map(String,long),Map(String,long))with ListCheckpointed{
private long numberRecords =0L;
@Override
flatMap(Tuple2(String,long)t,Collector collector){
//接入一条记录则进行统计，并输出
numberRecords +=1;
collector.collect(t.f0,numberRecords);
}
@Override
snapshotState(long checkpointId){
Collections.singletonList(numberRecords);
}
@Override
restoreState(List<long> list){
numberRecords =0L;
for(count <list){
//从状态中恢复numberRecords数据
numberRecords +=count
}
}
}

 

　　空洞的代码是没有感染力的，所以前面我铺垫了这么多，希望接下来的总结能对大家有所帮助。 　　 　　以上所有总结参考张利兵《flink实战，总结与分析》第五章。附：原文是用scala写的，因为楼主所在的公司用的java，所以将所有代码用java改写了一遍。如果有看着不方便的   朋友，可以去看原文嘿嘿。   　　作为刚接触flink的小白，本文只讲了状态的基本知识。后续可能会有如何存储状态以及flink状态机制优化。   　　欢迎大家不吝赐教。