引言
在流進行中，資料是連續不斷到來和處理的。每個任務在進行計算處理時，都可以基于目前資料直接轉換得到輸出結果；也可以依賴一些其他資料。這些由一個任務維護且用來計算輸出結果的所有資料就叫作這個任務的狀态。
一、什麼是狀态
1.1有狀态的算子
在 Flink 中，算子任務可以分為無狀态和有狀态兩種情況。
基本轉換算子，如 map、filter、flatMap， 計算時不依賴其他資料，就都屬于無狀态的算子。
常見的聚合算子、視窗算子都屬于有狀态的算子。
1.2狀态的管理
在傳統的事務型處理架構中，這種額外的狀态資料是儲存在資料庫中的。而對于實時流處理來說，這樣做需要頻繁讀寫外部資料庫，如果資料規模非常大肯定就達不到性能要求了。是以 Flink 的解決方案是，将狀态直接儲存在記憶體中來保證性能，并通過分布式擴充來提高吞吐量。
1.3狀态的分類
1. 托管狀态（Managed State）和原始狀态（Raw State）
托管狀态就是由 Flink 統一管理的，狀态的存儲通路、故障恢複和重組等一系列問題都由 Flink 實作，我們隻要調接口就可以。聚合、視窗等算子中内置的狀态，就都是托管狀态；我們也可以在富函數類（RichFunction）中通過上下文來自定義狀态，這些也都是托管狀态。
而原始狀态則是自定義的，相當于就是開辟了一塊記憶體，需要我們自己管理，實作狀态的序列化和故障恢複。隻有在遇到托管狀态無法實作的特殊需求時，我們才會考慮使用原始狀态；
2. 算子狀态（Operator State）和按鍵分區狀态（Keyed State）
我們又可以将托管狀态分為兩類：算子狀态和按鍵分區狀态。
| Operator State | Keyed State | |
| 适用算子類型 | 可用于所有算子: 常用于source, sink, 例如 FlinkKafkaConsumer | 隻能用于用于KeyedStream上的算子 |
| 狀态配置設定 | 一個算子的子任務對應一個狀态 | 一個Key對應一個State: 一個算子會處理多個Key, 則通路相應的多個State |
| 建立和通路方式 | 實作CheckpointedFunction或ListCheckpointed(已經過時)接口 | 重寫RichFunction, 通過裡面的RuntimeContext通路 |
| 橫向擴充 | 并發改變時有多重重寫配置設定方式可選: 均勻配置設定和合并後每個得到全量 | 并發改變, State随着Key在執行個體間遷移 |
| 支援的資料結構 | ListState, UnionListStste, BroadCastState | ValueState, ListState, MapState, ReduceState, AggregatingState |
Operator State的實際應用場景不如Keyed State多，它經常被用在Source或Sink等算子上，用來儲存流入資料的偏移量或對輸出資料做緩存，以保證Flink應用的Exactly-Once語義。無論是 Keyed State 還是 Operator State，它們都是在本地執行個體上維護的，也就是說每個并行子任務維護着對應的狀态，算子的子任務之間狀态不共享。
二、Flink的狀态和資料結構
2.1鍵控狀态
具有相同 key 的所有資料都會到通路相同的狀态，而不同 key 的狀态之間是彼此隔離的。
鍵控狀态是根據輸入資料流中定義的鍵（key）來維護和通路的。
Flink為每個鍵值維護一個狀态執行個體，并将具有相同鍵的所有資料，都分區到同一個算子任務中，這個任務會維護和處理這個key對應的狀态。當任務處理一條資料時，它會自動将狀态的通路範圍限定為目前資料的key。是以，具有相同key的所有資料都會通路相同的狀态。Keyed State很類似于一個分布式的key-value map資料結構，隻能用于KeyedStream（keyBy算子處理之後）。
鍵控狀态支援的資料結構
- ValueState<T>
- 儲存單個值。每個key有一個狀态值。設定使用 update(T)，擷取使用 T value()
- ListState<T>
- 儲存元素清單。
- 添加元素：add(T) addAll(List<T>)
- 擷取元素：Iterable<T> get()
- 覆寫所有元素：update(List<T>)
- ReducingState<T>
- 存儲單個值，表示把所有元素的聚合結果添加到狀态中。與ListState類似，但是當使用add(T)的時候ReducingState會使用指定的ReduceFunction進行聚合。
- AggregatingState<IN, OUT>
- 存儲單個值。與ReducingState類似，都是進行聚合。不同的是，AggregatingState的聚合結果和元素類型可以不一樣。
- MapState<UK, UV>
- 存儲鍵值對清單。
- 添加鍵值對：put(UK, UV) or putAll(Map<UK, UV>)
- 根據key擷取值：get(UK)
- 擷取所有：entries(), keys() and values()
- 檢測是否為空：isEmpty()
注意：
a) 所有的類型都有clear()，清空目前key的狀态
b) 這些狀态對象僅用于使用者與狀态進行互動
c) 狀态不是必須存儲到記憶體，也可以存儲在磁盤或者任意其他地方
d) 從狀态擷取的值與輸入元素的key相關
2.2算子狀态
除按鍵分區狀态之外，另一大類受控狀态就是算子狀态。從某種意義上說，算子狀态是更底層的狀态類型，因為它隻針對目前算子并行任務有效，不需要考慮不同 key 的隔離。算子狀态功能不如按鍵分區狀态豐富，應用場景較少，經常被用在Source或Sink等算子上，用來儲存流入資料的偏移量或對輸出資料做緩存，Kafka consumer 每個并行執行個體維護了 topic partitions 和偏移量的map作為它的算子狀态，以保證Flink應用的Exactly-Once語義。
Flink為算子狀态提供三種基本資料結構：
- 清單狀态（List state）
- 将狀态表示為一組資料的清單
- 聯合清單狀态（Union list state）
- 也是将狀态表示為資料的清單。它與正常清單狀态的差別在于，在發生故障時，或者從儲存點（savepoint）啟動應用程式時如何恢複。
- 一種是均勻配置設定(List state)，另外一種是将所有 State 合并為全量 State 再分發給每個執行個體(Union list state)。
- 廣播狀态（Broadcast state）
- 是一種特殊的算子狀态。如果一個算子有多項任務，而它的每項任務狀态又都相同，那麼這種特殊情況最适合應用廣播狀态。
三、狀态持久化和狀态後端
Flink 的狀态管理機制中，很重要的一個功能就是對狀态進行持久化（persistence）儲存，這樣就可以在發生故障後進行重新開機恢複。Flink 對狀态進行持久化的方式，就是将目前所有分布式狀态進行“快照”儲存，寫入一個“檢查點”（checkpoint）或者儲存點（savepoint）儲存到外部存儲系統中。具體的存儲媒體，一般是分布式檔案系統（distributed file system）。
3.1檢查點
有狀态流應用中的檢查點（checkpoint），其實就是所有任務的狀态在某個時間點的一個快照（一份拷貝）。簡單來講，就是一次“存盤”，讓我們之前處理資料的進度不要丢掉。在一個流應用程式運作時，Flink 會定期儲存檢查點，在檢查點中會記錄每個算子的 id 和狀态；如果發生故障，Flink 就會用最近一次成功儲存的檢查點來恢複應用的狀态，重新啟動處理流程， 就如同“讀檔”一樣。
3.2狀态後端
在 Flink 中，狀态的存儲、通路以及維護，都是由一個可插拔的元件決定的，這個元件就叫作狀态後端（state backend）。狀态後端主要負責兩件事：一是本地的狀态管理，二是将檢查點（checkpoint）寫入遠端的持久化存儲。
1. 狀态後端的分類
Flink 中提供了兩類不同的狀态後端，一種是“哈希表狀态後端”（HashMapStateBackend），另一種是“内嵌 RocksDB 狀态後端”（EmbeddedRocksDBStateBackend）。如果沒有特别配置，系統預設的狀态後端是HashMapStateBackend。
- 哈希表狀态後端（HashMapStateBackend）
- HashMapStateBackend 是将本地狀态全部放入記憶體的，這樣可以獲得最快的讀寫速度，使計算性能達到最佳；代價則是記憶體的占用。它适用于具有大狀态、長視窗、大鍵值狀态的作業， 對所有高可用性設定也是有效的。
- 對于檢查點的儲存，一般是放在持久化的分布式檔案系統（file system）中，也可以通過配置“檢查點存儲”（CheckpointStorage）來另外指定。
- 内嵌RocksDB 狀态後端（EmbeddedRocksDBStateBackend）
- EmbeddedRocksDBStateBackend會将進行中的資料全部放入 RocksDB 資料庫中，資料被存儲為序列化的位元組數組，讀寫操作需要序列化/反序列化，是以狀态的通路性能要差一些。EmbeddedRocksDBStateBackend始終執行的是異步快照，也就是不會因為儲存檢查點而阻塞資料的處理；而且它還提供了增量式儲存檢查點的機制，這在很多情況下可以大大提升儲存效率。對于檢查點，同樣會寫入到遠端的持久化檔案系統中。由于它會把狀态資料落盤，而且支援增量化的檢查點，是以在狀态非常大、視窗非常長、鍵/值狀态很大的應用場景中是一個好選擇，同樣對所有高可用性設定有效。
2. 如何選擇正确的狀态後端
- HashMap 和 RocksDB 兩種狀态後端最大的差別，就在于本地狀态存放在哪裡：前者是記憶體，後者是 RocksDB。在實際應用中，選擇哪種狀态後端，主要是需要根據業務需求在處理性能和應用的擴充性上做一個選擇。在工作中，RocksDB讀寫性能可能不如HashMap，但是由于RocksDb存儲在硬碟上，是以RocksDb往往更受公司的歡迎。
總結
Flink的狀态，我們可以當成類似于redis去了解，狀态的資料類型也就是存儲資料的資料類型。鍵控狀态我們使用的會比較多一點，另外多并行度的情況，每個并行度下的狀态值都不是共享的。狀态後端的話，資料量如果比較大，建議使用RocksDB。