五、flink--state狀態管理機制

一、flink狀態原理

1.1 什么是flink中的狀態？為什么需要狀態管理？

? flink運行計算任務的過程中，會有很多中間處理過程。在整個任務運行的過程中，中間存在著多個臨時狀態，比如說某些數據正在執行一個operator，但是只處理了一半數據，另外一般還沒來得及處理，這也是一個狀態。
? 假設運行過程中，由于某些原因，任務掛掉了，或者flink中的多個task中的一個task掛掉了，那么它在內存中的狀態都會丟失，如果這時候我們沒有存儲中間計算的狀態，那么就意味著重啟這個計算任務時，需要從頭開始將原來處理過的數據重新計算一遍。如果存儲了中間狀態，就可以恢復到中間狀態，并從該狀態開始繼續執行任務。這就是狀態管理的意義。所以需要一種機制去保存記錄執行過程中的中間狀態，這種機制就是狀態管理機制。

1.2 flink中狀態的分類

flink中包括兩種基礎狀態：keyed state（keyed狀態）和operator state（operator狀態）

1.2.1 keyed狀態

keyed狀態總是與key一起，且只能用在keyedStream中。這個狀態是跟特定的key綁定的，對KeyedStream流上的每一個key，可能都對應一個state。唯一組合成（operator--key，state）的形式。

1.2.2 operator狀態

與Keyed State不同，Operator State跟一個特定operator的一個并發實例綁定，整個operator只對應一個state。相比較而言，在一個operator上，可能會有很多個key，從而對應多個keyed state。而且operator state可以應用于非keyed stream中。

舉例來說，Flink中的Kafka Connector，就使用了operator state。它會在每個connector實例中，保存該實例中消費topic的所有(partition, offset)映射。

1.3 state的存在形式

所有state的存在形式有兩種：managed（委托管理）和raw（原始）。
托管方式就是狀態管理由flink提供的框架進行管理，通過flink狀態管理框架提供的接口，來更新和管理狀態的值。這里面包括用于存儲狀態數據的數據結構，現成的包裝類等。

原始方式就是由用戶自行管理狀態具體的數據結構，框架在做checkpoint的時候（checkpoint是flink進行狀態數據持久化存儲的機制），使用byte[]來讀寫狀態內容，對其內部數據結構一無所知。

通常在DataStream上的狀態推薦使用托管的狀態，當實現一個用戶自定義的operator時，會使用到原始狀態。一般來說，托管狀態用的比較多。

1.4 managed 方式提供的接口

flink提供了很多存儲state數據的類，類繼承圖如下：

? 圖 1.4 flink--狀態管理類繼承圖

可用于存儲狀態數據的類如下：

ValueState<T>*這保留了一個可以更新和檢索的值。即類型為T的單值狀態。這個狀態與對應的key綁定，是最簡單的狀態了。它可以通過update方法更新狀態值，通過value()方法獲取狀態值。

ListState<T>*這保存了一個元素列表，即key上的狀態值為一個列表。可以追加元素并檢索Iterable所有當前存儲的元素。可以通過add方法往列表中附加值；也可以通過get()方法返回一個Iterable<T>來遍歷狀態值。

ReducingState<T>*這保留了一個值，該值表示添加到狀態的所有值的聚合。接口類似于ListState，但是添加的元素使用add(T)。每次調用add方法添加值的時候，會調用reduceFunction，最后合并到一個單一的狀態值。

AggregatingState<IN, OUT>*這保留了一個值，該值表示添加到狀態的所有值的聚合。和ReducingState不同的是，聚合類型可能與添加到狀態的元素類型不同

FoldingState<T, ACC>*這保留了一個值，該值表示添加到狀態的所有值的聚合。違背ReducingState，聚合類型可能與添加到狀態的元素類型不同。接口類似于ListState但是添加的元素使用add(T)使用指定的FoldFunction.這個基本棄用，請用AggregatingState代替

MapState<UK, UV>*它保存了一個映射列表。可以將鍵值對放入狀態并檢索Iterable所有當前存儲的映射。映射使用put(UK, UV)或putAll(Map<UK, UV>)添加元素。使用get(UK)獲取元素。映射、鍵和值的可迭代視圖可以分別使用entries(), keys()和values()

需要注意的是，以上所述的State對象，僅僅用于與狀態進行交互（更新、刪除、清空等），而真正的狀態值，有可能是存在內存、磁盤、或者其他分布式存儲系統中。相當于我們只是持有了這個狀態的句柄(state handle)。

接下來看下，我們如何得到這個狀態句柄。Flink通過StateDescriptor來定義一個狀態。這是一個抽象類，內部定義了狀態名稱、類型、序列化器等基礎信息。與上面的狀態類型對應。如下：

ValueStateDescriptor
ListStateDescriptor
ReducingStateDescriptor
FoldingStateDescriptor
AggregatingStateDescriptor
MapStateDescriptor

二、使用flink狀態管理的方式

2.1 使用狀態管理基本流程

以keyed state為例，
1、首先，普通Function接口是不支持狀態管理的，也就是一般故障的情況下，狀態并沒有保存下來，后面需要將所有數據進行重新計算。如果需要支持狀態管理，那么我們需要繼承實現 RichFunction類。基本常用的function，flink都再封裝了對應的RichFunction接口給我們使用，比如普通function中的MapFunction，對應的RichFunction抽象類為RichMapFunction。命名方式對應關系很簡單，基本就是 xxxFunciotn -->RichxxxFunction。

2、接著，需要在覆蓋實現RichFunction中的對應的算子方法（如map、flatMap等），里面需要實現算子業務邏輯，并將對keyed state進行更新、管理。然后還要重寫open方式，用于獲取狀態句柄。

2.2 使用keyed state例子

下面使用ValueState為例，實現RichFlatMapFunction接口：

public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {

    /**
     * ValueState狀態句柄. 第一個值為count，第二個值為sum。
     */
    private transient ValueState<Tuple2<Long, Long>> sum;

    @Override
    public void flatMap(Tuple2<Long, Long> input, Collector<Tuple2<Long, Long>> out) throws Exception {
        // 獲取當前狀態值
        Tuple2<Long, Long> currentSum = sum.value();

        // 更新
        currentSum.f0 += 1;
        currentSum.f1 += input.f1;

        // 更新狀態值
        sum.update(currentSum);

        // 如果count >=2 清空狀態值，重新計算
        if (currentSum.f0 >= 2) {
            out.collect(new Tuple2<>(input.f0, currentSum.f1 / currentSum.f0));
            sum.clear();
        }
    }

    @Override
    public void open(Configuration config) {
        ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
                new ValueStateDescriptor<>(
                        "average", // 狀態名稱
                        TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {}), // 當個狀態數據的類型，這里是tuple,也就是元祖
                        Tuple2.of(0L, 0L)); // 狀態默認值
        //獲取狀態句柄
        sum = getRuntimeContext().getState(descriptor);
    }
}

// 主程序
env.fromElements(Tuple2.of(1L, 3L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L), Tuple2.of(1L, 4L), Tuple2.of(1L, 2L))
        .keyBy(0)
        .flatMap(new CountWindowAverage())
        .print();

// the printed output will be (1,4) and (1,5)

三、狀態管理源碼分析

3.1 獲取state操作句柄源碼分析

以2.2 中的例子為例，看狀態句柄是如何獲取的。
首先從這里入手：

sum = getRuntimeContext().getState(descriptor);

首先RichFlatMapFunction類中沒有這個方法，再到父類AbstractRichFunction，如下：

public abstract class AbstractRichFunction implements RichFunction, Serializable {
    private static final long serialVersionUID = 1L;
    private transient RuntimeContext runtimeContext;

    public AbstractRichFunction() {
    }

    public void setRuntimeContext(RuntimeContext t) {
        this.runtimeContext = t;
    }

    //返回RuntimeContext類的對象
    public RuntimeContext getRuntimeContext() {
        if (this.runtimeContext != null) {
            return this.runtimeContext;
        } else {
            throw new IllegalStateException("The runtime context has not been initialized.");
        }
    }
    ...........................
}

getRuntimeContext()這個方法在這里實際上是返回了StreamingRuntimeContext這個子類對象。所以調用了RichFlatMapFunction.getRuntimeContext().getState方法，最終會調用StreamingRuntimeContext.getState方法：

/* StreamingRuntimeContext */
public class StreamingRuntimeContext extends AbstractRuntimeUDFContext {
    ...........
    public <T> ValueState<T> getState(ValueStateDescriptor<T> stateProperties) {
        //這里獲取到KeyedStateStore對象
        KeyedStateStore keyedStateStore = this.checkPreconditionsAndGetKeyedStateStore(stateProperties);
        stateProperties.initializeSerializerUnlessSet(this.getExecutionConfig());
        return keyedStateStore.getState(stateProperties);
    }      
    ...............
}

接著看this.checkPreconditionsAndGetKeyedStateStore，

private KeyedStateStore checkPreconditionsAndGetKeyedStateStore(StateDescriptor<?, ?> stateDescriptor) {
        Preconditions.checkNotNull(stateDescriptor, "The state properties must not be null");
        KeyedStateStore keyedStateStore = this.operator.getKeyedStateStore();
        Preconditions.checkNotNull(keyedStateStore, "Keyed state can only be used on a 'keyed stream', i.e., after a 'keyBy()' operation.");
        return keyedStateStore;
    }

再接著看 this.operator.getKeyedStateStore();，

/*AbstractStreamOperator*/
.....
public KeyedStateStore getKeyedStateStore() {
        return this.keyedStateStore;
    }
.....

返回了AbstractStreamOperator.keyedStateStore變量。這個變量的初始化在AbstractStreamOperator.initializeState方法中：

public final void initializeState() throws Exception {
        TypeSerializer<?> keySerializer = this.config.getStateKeySerializer(this.getUserCodeClassloader());
        StreamTask<?, ?> containingTask = (StreamTask)Preconditions.checkNotNull(this.getContainingTask());
        CloseableRegistry streamTaskCloseableRegistry = (CloseableRegistry)Preconditions.checkNotNull(containingTask.getCancelables());
        //創建 StreamTaskStateInitializerImpl 對象
        StreamTaskStateInitializer streamTaskStateManager = (StreamTaskStateInitializer)Preconditions.checkNotNull(containingTask.createStreamTaskStateInitializer());
        //這里創建一個當前operator的state context對象
        StreamOperatorStateContext context = streamTaskStateManager.streamOperatorStateContext(this.getOperatorID(), this.getClass().getSimpleName(), this, keySerializer, streamTaskCloseableRegistry);
        //通過state context這個上下文對象獲取keyed state和operator state的 backend配置
        this.operatorStateBackend = context.operatorStateBackend();
        this.keyedStateBackend = context.keyedStateBackend();
        //初始化keyedStateStore，將從配置中創建的statebackend封裝到默認的statebackend
        if (this.keyedStateBackend != null) {
            this.keyedStateStore = new DefaultKeyedStateStore(this.keyedStateBackend, this.getExecutionConfig());
        }

        this.timeServiceManager = context.internalTimerServiceManager();
        CloseableIterable<KeyGroupStatePartitionStreamProvider> keyedStateInputs = context.rawKeyedStateInputs();
        CloseableIterable operatorStateInputs = context.rawOperatorStateInputs();

        try {
            StateInitializationContext initializationContext = new StateInitializationContextImpl(context.isRestored(), this.operatorStateBackend, this.keyedStateStore, keyedStateInputs, operatorStateInputs);
            this.initializeState(initializationContext);
        } finally {
            closeFromRegistry(operatorStateInputs, streamTaskCloseableRegistry);
            closeFromRegistry(keyedStateInputs, streamTaskCloseableRegistry);
        }

    }

在containingTask.createStreamTaskStateInitializer()，也就是StreamTask.createStreamTaskStateInitializer()這里有初始化操作，看看

public StreamTaskStateInitializer createStreamTaskStateInitializer() {
        return new StreamTaskStateInitializerImpl(this.getEnvironment(), this.stateBackend, this.timerService);
    }

創建了一個StreamTaskStateInitializerImpl對象，里面傳入了 this.stateBackend這個參數，估計就是獲取用戶指定的stateBackend類型，追蹤this.stateBackend這個參數在哪初始化的，找到了StreamTask.invoke這個方法，如下：

public final void invoke() throws Exception {
        ..........
            this.stateBackend = this.createStateBackend();
        ............
            }

調用了this.createStateBackend()，即StreamTask.createStateBackend()繼續看

private StateBackend createStateBackend() throws Exception {
    //從application中獲取已經創建的statebackend object
        StateBackend fromApplication = this.configuration.getStateBackend(this.getUserCodeClassLoader());
    //加載指定的 statebackend類，并返回statebackend對象
        return StateBackendLoader.fromApplicationOrConfigOrDefault(fromApplication, this.getEnvironment().getTaskManagerInfo().getConfiguration(), this.getUserCodeClassLoader(), LOG);
    }

繼續看this.configuration.getStateBackend, 即StreamConfing.getStateBackend

public StateBackend getStateBackend(ClassLoader cl) {
        try {
            //從application的config中獲取"statebackend"這個key對應的object，也就是已創建的backend對象。
            return (StateBackend)InstantiationUtil.readObjectFromConfig(this.config, "statebackend", cl);
        } catch (Exception var3) {
            throw new StreamTaskException("Could not instantiate statehandle provider.", var3);
        }
    }

從application的config中獲取"statebackend"這個key對應的object，也就是已創建的backend對象。如果application不存在，則從配置中讀取應該使用的backend類型，然后加載。看這個方法StateBackendLoader.fromApplicationOrConfigOrDefault。這個方法就是從config配置中讀取“state.backend”這個配置項值，然后加載對應的backend。

到最后，要記住this.keyedStateStore = new DefaultKeyedStateStore(this.keyedStateBackend, this.getExecutionConfig());這個是 this.keyedStateStore封裝后的對象。
接著回到StreamingRuntimeContext.getState中，看到 this.keyedStateStore.getState，這一行，調用的是DefaultKeyedStateStore.getState,其實就再次進行初始化而已，這里不展開。

3.2 獲取和更新狀態值源碼分析

首先，我們這里使用的是ValueState，它的直接實現子類是HeapValueState。它的存儲數據結構，在它的父類AbstractHeapState中，以StateTable<K, N, SV> stateTable的形式存在的，其中K代表Key的類型，N代表state的namespace（這樣屬于不同namespace的state可以重名），SV代表state value的類型。