本文主要內容如下：

1. 有狀態的流資料處理；
2. Flink中的狀態介面；
3. 狀態管理和容錯機制實現；

一.有狀態的流資料處理

1.1.什麼是有狀態的計算

計算任務的結果不僅僅依賴於輸入，還依賴於它的當前狀態，其實大多數的計算都是有狀態的計算。 比如wordcount,給一些word,其計算它的count,這是一個很常見的業務場景。count做為輸出，在計算的過程中要不斷的把輸入累加到count上去，那麼count就是一個state。

1.2.傳統的流計算系統缺少對於程式狀態的有效支援

1. 狀態資料的儲存和訪問；
2. 狀態資料的備份和恢復；
3. 狀態資料的劃分和動態擴容；
   

在傳統的批處理中，資料是劃分為塊分片去完成的，然後每一個Task去處理一個分片。當分片執行完成後，把輸出聚合起來就是最終的結果。在這個過程當中，對於state的需求還是比較小的。

對於流計算而言，對State有非常高的要求，因為在流系統中輸入是一個無限制的流，會執行很長一段時間，甚至執行幾天或者幾個月都不會停機。在這個過程當中，就需要將狀態資料很好的管理起來。很不幸的是，在傳統的流計算系統中，對狀態管理支援並不是很完善。比如storm, 沒有任何程式狀態的支援，一種可選的方案是storm+hbase這樣的方式去實現，把這狀態資料存放在Hbase中，計算的時候再次從Hbase讀取狀態資料，做更新在寫入進去。這樣就會有如下幾個問題

流計算系統的任務和Hbase的資料儲存有可能不在同一臺機器上，導致效能會很差。這樣經常會做遠端的訪問，走網路和儲存；

備份和恢復是比較困難，因為Hbase是沒有回滾的，要做到Exactly onces很困難。在分散式環境下，如果程式出現故障，只能重啟Storm，那麼Hbase的資料也就無法回滾到之前的狀態。比如廣告計費的這種場景，Storm+Hbase是是行不通的，出現的問題是錢可能就會多算，解決以上的辦法是Storm+mysql，通過mysql的回滾解決一致性的問題。但是架構會變得非常複雜。效能也會很差，要commit確保資料的一致性。

對於storm而言狀態資料的劃分和動態擴容也是非常難做，一個很嚴重的問題是所有使用者都會在strom上重複的做這些工作，比如搜尋，廣告都要在做一遍，由此限制了部門的業務發展。

1.3.Flink豐富的狀態訪問和高效的容錯機制

Flink在最早設計的時候就意識到了這個問題，並提供了豐富的狀態訪問和容錯機制。如下圖所示：

二．Flink中的狀態管理

2.1.按照資料的劃分和擴張方式，Flink中大致分為2類

1. Keyed States
2. Operator States
   

2.1.1.Keyed States

Keyed States的使用:

Flink也提供了Keyed States多種資料結構型別:

Keyed States的動態擴容:

2.1.2.Operator State

Operator States的使用:

Operator States的資料結構不像Keyed States豐富，現在只支援List。
Operator States多種擴充套件方式:

Operator States的動態擴充套件是非常靈活的，現提供了3種擴充套件，下面分別介紹：

1. ListState:併發度在改變的時候，會將併發上的每個List都取出，然後把這些List合併到一個新的List,然後根據元素的個數在均勻分配給新的Task;

2. UnionListState:相比於ListState更加靈活，把劃分的方式交給使用者去做，當改變併發的時候，會將原來的List拼接起來。然後不做劃分，直接交給使用者；

3. BroadcastState:如大表和小表做Join時，小表可以直接廣播給大表的分割槽，在每個併發上的資料都是完全一致的。做的更新也相同，當改變併發的時候，把這些資料COPY到新的Task即可

以上是Flink Operator States提供的3種擴充套件方式，使用者可以根據自己的需求做選擇。

2.2. 使用Checkpoint提高程式的可靠性

使用者可以根據的程式裡面的配置將checkpoint開啟，給定一個時間間隔後，框架會按照時間間隔給程式的狀態進行備份。當發生故障時，Flink會將所有Task的狀態一起恢復到Checkpoint的狀態。從哪個位置開始重新執行。

Flink也提供了多種正確性的保障，包括：

1. at least once
2. exactly once

2.3. 備份為儲存在State中的程式狀態資料

Flink也提供了一套機制，允許把這些狀態放到記憶體當中。做Checkpoint的時候，由Flink去完成恢復。

2.4. 從已停止作業的執行狀態中恢復

當元件升級的時候，需要停止當前作業。這個時候需要從之前停止的作業當中恢復，Flink提供了2種機制恢復作業:

1. Savepoint:是一種特殊的checkpoint，只不過不像checkpoint定期的從系統中去觸發的，它是使用者通過命令觸發，儲存格式和checkpoint也是不相同的，會將資料按照一個標準的格式儲存，不管配置什麼樣，Flink都會從這個checkpoint恢復，是用來做版本升級一個非常好的工具；

2. External Checkpoint：對已有checkpoint的一種擴充套件，就是說做完一次內部的一次Checkpoint後，還會在使用者給定的一個目錄中，多儲存一份checkpoint的資料；

三．狀態管理和容錯機制實現

下面介紹一下狀態管理和容錯機制實現方式，Flink提供了3種不同的StateBackend

1. MemoryStateBackend
2. FsStateBackend
3. RockDBStateBackend
   

使用者可以根據自己的需求選擇，如果資料量較小，可以存放到MemoryStateBackend和FsStateBackend中，如果資料量較大，可以放到RockDB中。

下面介紹HeapKeyedStateBackend和RockDBKeyedStateBackend:
第一，HeapKeyedStateBackend

第二，RockDBKeyedStateBackend

Checkpoint的執行流程
Checkpoint的執行流程是按照Chandy-Lamport演算法實現的。

Checkpoint Barrier的對齊

全量Checkpoint
全量Checkpoint會在每個節點做備份資料時，只需要將資料都便利一遍，然後寫到外部儲存中，這種情況會影響備份效能。在此基礎上做了優化。

RockDB的增量Checkpoint
RockDB的資料會更新到記憶體，當記憶體滿時，會寫入到磁碟中。增量的機制會將新產生的檔案COPY持久化中，而之前產生的檔案就不需要COPY到持久化中去了。通過這種方式減少COPY的資料量，並提高效能。