Flink狀態專題：keyed state和Operator state
正好最近公司有個需求，要用到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
（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狀態機制優化。 歡迎大家不吝賜教。