寫在前面

State是指流計算過程中計算節(jié)點的中間計算結(jié)果或元數(shù)據(jù)屬性，比如 在aggregation過程中要在state中記錄中間聚合結(jié)果，比如 Apache Kafka 作為數(shù)據(jù)源時候，我們也要記錄已經(jīng)讀取記錄的offset，這些State數(shù)據(jù)在計算過程中會進行持久化(插入或更新)。本文將詳細介紹一下Flink State，通過本文，你可以了解到：

• State分類
• 什么是狀態(tài)后端(state backend)
• State對擴縮容的處理

感謝關(guān)注，希望本文對你有所幫助。

State分類

Flink 中的狀態(tài)分為兩種主要類型：Keyed State 和 Operator State。

Keyed State

• 概念：Keyed State 是和鍵（key）相關(guān)聯(lián)的狀態(tài)。在 Flink 的 Keyed Streams 上進行有狀態(tài)操作時（例如在使用 keyBy 方法后），每個 key 都會有自己的狀態(tài)實例，這個狀態(tài)是獨立的，即每個 key 的狀態(tài)對于其他 keys 不可見。
• 用法：Keyed State 常用于需要按 key 進行分區(qū)處理的情況，如聚合計算（sum、min、max）、窗口操作和其他需要按 key 維護和更新狀態(tài)的計算。在 SQL 語句中，Keyed State 對應的就是通過 GroupBy 或 PartitionBy 所定義的字段分組。
• 數(shù)據(jù)結(jié)構(gòu)：Keyed State 底層通常是基于哈希表的實現(xiàn)，確保每個 key 都能快速地找到對應的狀態(tài)。這種狀態(tài)通常存儲在 Keyed State 后端中，可以是內(nèi)存中，也可以是 RocksDB 這種本地存儲。

Operator State

• 概念：Operator State 與特定的操作符實例（Task）相關(guān)聯(lián)，而不是和特定的 key 關(guān)聯(lián)。每個操作符實例維護自己的狀態(tài)，所有的 Operator State 實例對于同一操作符是可見的。
• 用法：Operator State 通常用于記錄源（Source）和接收器（Sink）的相關(guān)狀態(tài)，或者用于需要操作符級別聚合的場合。例如，一個 Source Connector 可能會使用 Operator State 來記錄已經(jīng)讀取的數(shù)據(jù)源的 offset。
• 實現(xiàn)：Flink 提供了幾種不同的 Operator State 類型，包括列表狀態(tài)（ListState）、聯(lián)合列表狀態(tài)（UnionListState）、廣播狀態(tài)（BroadcastState）等。這些狀態(tài)通常存儲在 Operator State 后端中，可以是內(nèi)存中，也可以是持久化存儲。

值得注意的是：

在 Flink 的 Table API 或 SQL API 中，對于內(nèi)部的 GroupBy/PartitionBy 操作，F(xiàn)link 會自動管理 Keyed State。而對于 Source Connector 記錄 offset 這樣的操作，通常是在底層的 DataStream API 中實現(xiàn)的，可能直接使用 Operator State 來管理。例如，F(xiàn)link Kafka Consumer 會使用 Operator State 來存儲 Kafka 主題的分區(qū) offset，以便在發(fā)生故障時能夠從上次成功的檢查點恢復。

什么是狀態(tài)后端(state backend)

State的具體存儲、訪問和維護是由**狀態(tài)后端(state backend)**決定的。狀態(tài)后端主要負責兩件事情：

• 本地狀態(tài)管理
• 將狀態(tài)以checkpoint的形式寫入遠程存儲

Flink提供了三種狀態(tài)后端：

MemoryStateBackend（內(nèi)存狀態(tài)后端）

• 存儲：狀態(tài)存儲在 TaskManager 的 JVM 堆內(nèi)存上。生成checkpoint時，*MemoryStateBackend會將狀態(tài)發(fā)送至JobManager并保存到它的堆內(nèi)存中。
• 使用場景：適用于小規(guī)模狀態(tài)或本地測試，因為它將所有狀態(tài)作為序列化數(shù)據(jù)保存在 JVM 堆上。如果 TaskManager 發(fā)生故障，狀態(tài)會丟失。
• 性能：由于狀態(tài)是直接存儲在內(nèi)存中的，所以訪問速度很快。
• 限制：狀態(tài)大小受限于 TaskManager 可用內(nèi)存。大規(guī)模狀態(tài)可能導致內(nèi)存溢出錯誤。

FsStateBackend（文件系統(tǒng)狀態(tài)后端）

• 存儲：狀態(tài)存儲在 TaskManager 的 JVM 堆內(nèi)存中（作為緩存），但在檢查點（checkpoint）時，會持久化到配置的文件系統(tǒng)（如 HDFS）中。
• 使用場景：適用于需要持久化狀態(tài)以避免數(shù)據(jù)丟失的場景。在發(fā)生故障時，F(xiàn)link 作業(yè)可以從文件系統(tǒng)中的檢查點恢復狀態(tài)。
• 性能：由于狀態(tài)在內(nèi)存中進行操作，并在檢查點時異步寫入文件系統(tǒng)，因此可以提供較快的狀態(tài)訪問速度，但可能受文件系統(tǒng)性能的限制。
• 限制：內(nèi)存中的狀態(tài)大小仍然受限于 TaskManager 可用內(nèi)存，但由于檢查點數(shù)據(jù)被寫入到更穩(wěn)定的文件系統(tǒng)，因此可以支持更大的狀態(tài)。

RocksDBStateBackend（RocksDB 狀態(tài)后端）

RocksDB是一個嵌入式鍵值存儲(key-value store)，它可以將數(shù)據(jù)保存到本地磁盤上，為了從RocksDB中讀寫數(shù)據(jù)，系統(tǒng)需要對數(shù)據(jù)進行序列化和反序列化。

• 存儲：狀態(tài)存儲在本地磁盤上的 RocksDB 數(shù)據(jù)庫中，檢查點數(shù)據(jù)會持久化到配置的文件系統(tǒng)中。
• 使用場景：適用于大規(guī)模狀態(tài)管理的場景。由于 RocksDB 是一個優(yōu)化的鍵值存儲，因此可以有效地管理大量狀態(tài)數(shù)據(jù)。
• 性能：狀態(tài)訪問速度可能比內(nèi)存狀態(tài)后端慢(磁盤讀寫以及序列化和反序列化對象的開銷)，但 RocksDB 提供了針對大量狀態(tài)數(shù)據(jù)的優(yōu)化。
• 限制：對本地磁盤空間有需求，但由于狀態(tài)是在本地磁盤上操作，因此可以支持非常大的狀態(tài)。

在選擇狀態(tài)后端時，需要考慮應用的狀態(tài)大小、恢復速度、持久性和部署環(huán)境。對于生產(chǎn)環(huán)境，通常推薦使用 RocksDBStateBackend，因為它能夠提供良好的擴展性和容錯性。

State對擴縮容的處理

Operator State 的擴容處理

在 Apache Flink 中，對于有狀態(tài)的流處理作業(yè)，當作業(yè)進行擴容（scaling out）或縮容（scaling in）時，即增加或減少并行子任務(wù)的數(shù)量時，F(xiàn)link 需要重新分配 OperatorState。這個過程稱為狀態(tài)重分配（state redistribution）。

對于 Operator State 的擴容處理，F(xiàn)link 提供了不同的重分配模式來處理狀態(tài)：

ListState

對于 ListState 類型的 Operator State，如果流任務(wù)的并行度從 N 增加到 M，F(xiàn)link 會將每個并行實例的狀態(tài)分成 M 份，然后將這些分片分配給新的并行實例。如果并行度減少，則相反，狀態(tài)將會聚合起來。

圖片

擴容時：

• 假設(shè)原來有 2 個并行實例，每個實例有自己的 ListState。
• 擴容到 3 個并行實例。
• Flink 會將每個原來的 ListState 平均分成 3 份。
• 新的 3 個并行實例每個都會接收一份來自每個原始 ListState 的數(shù)據(jù)。

縮容時：

• 假設(shè)原來有 3個并行實例。
• 縮容到 1 個并行實例。
• 現(xiàn)有的狀態(tài)將會被聚合，確保新的 1 個實例完整地包含原始狀態(tài)的全部數(shù)據(jù)。

BroadcastState

BroadcastState 的數(shù)據(jù)在擴容或縮容時會被復制到所有的并行實例中。由于 BroadcastState 是以廣播的方式存儲數(shù)據(jù)，所有并行實例的狀態(tài)都是相同的。

圖片

UnionListState

對于 UnionListState 類型的 Operator State，在擴容或縮容時，狀態(tài)的每個元素將保持不變，原始狀態(tài)的所有元素將被統(tǒng)一地分發(fā)到新的并行實例中。這意味著每個元素僅分配給一個并行實例，但所有并行實例的狀態(tài)的并集會包括所有原始狀態(tài)的元素。隨后由任務(wù)自己決定哪些條目該保留，哪些該丟棄。

圖片

思考：Source的擴容（并發(fā)數(shù)）是否可以超過Source物理存儲的partition數(shù)量呢？

在使用像 Apache Kafka 這樣的消息隊列作為數(shù)據(jù)源（Source）時，消息隊列中的數(shù)據(jù)被劃分為多個分區(qū)（partitions）。這種設(shè)計主要是為了支持數(shù)據(jù)的并行處理以及提高吞吐量。在使用 Flink 或類似的流處理框架時，一個常見的做法是將每個分區(qū)分配給一個并行的 Source 實例（也稱為 Source Task 或 Source Operator）進行處理。

如果嘗試將 Source 的并行度（并發(fā)數(shù)）設(shè)置得比物理存儲（比如 Kafka 主題）的分區(qū)數(shù)量還要高，那么將會有一些并行實例分配不到任何分區(qū)，因為分區(qū)的數(shù)量是固定的，且每個分區(qū)只能被一個并行實例消費（至少在 Flink 的默認設(shè)置下是這樣）。這會導致資源浪費，因為超出分區(qū)數(shù)量的那部分并行實例不會做任何實際的數(shù)據(jù)處理工作，但仍然占用系統(tǒng)資源。

因此，在設(shè)置 Source 的并行度時，通常的最佳實踐是：

• 確保 Source 的并行度不超過其對應物理存儲（如 Kafka 主題）的分區(qū)數(shù)量。

如果需要增加并行度以提高處理能力，相應地也需要增加物理存儲的分區(qū)數(shù)量。對于 Kafka 來說，可以通過修改主題的分區(qū)配置來實現(xiàn)。
對于 Apache Flink，如果使用的是 Flink Kafka Connector，并且嘗試將并行度設(shè)置得比 Kafka 主題的分區(qū)數(shù)量還要高，F(xiàn)link 會在作業(yè)啟動時進行檢查。如果發(fā)現(xiàn)這種配置不匹配的情況，F(xiàn)link 會拋出異常并終止作業(yè)啟動，以避免資源浪費和潛在的配置錯誤。這種設(shè)計選擇確保了資源的有效利用和處理能力的合理分配，同時也避免了由于配置錯誤而導致的潛在問題。

KeyedState對擴容的處理

• 什么是Key-Groups

KeyedState的算子在擴容時會根據(jù)新的任務(wù)數(shù)量對key進行重分區(qū)，為了降低狀態(tài)在不同任務(wù)之間遷移的成本，F(xiàn)link不會單獨對key進行在分配，而是會把所有的鍵值分別存到不同的key-group中，每個key-group都包含了部分鍵值對。一個key-group是State分配的原子單位。

• 什么決定Key-Groups的個數(shù)

key-group的數(shù)量在job啟動前必須是確定的且運行中不能改變。由于key-group是state分配的原子單位，而每個operator并行實例至少包含一個key-group，因此operator的最大并行度不能超過設(shè)定的key-group的個數(shù)，那么在Flink的內(nèi)部實現(xiàn)上key-group的數(shù)量就是最大并行度的值。

• 如何決定key屬于哪個Key-Group

為了決定一個key屬于哪個Key-Group，通常會采用一種叫做一致性哈希（Consistent Hashing）的算法。一致性哈希算法的基本思想是將所有的Key和所有的Key-Group都映射到同一個哈希環(huán)上。對每個Key進行哈希運算得到一個哈希值，然后在哈希環(huán)上找到一個順時針方向最近的Key-Group，這個Key就屬于這個Key-Group。即：Key到指定的key-group的邏輯是利用key的hashCode和maxParallelism取余操作的來分配的。

如下圖當parallelism=2,maxParallelism=10的情況下流上key與key-group的對應關(guān)系如下圖所示：

圖片

如上圖key(a)的hashCode是97，與最大并發(fā)10取余后是7，被分配到了KG-7中，流上每個event都會分配到KG-0至KG-9其中一個Key-Group中。

上面的Stateful Operation節(jié)點的最大并行度maxParallelism的值是10，也就是我們一共有10個Key-Group，當我們并發(fā)是2的時候和并發(fā)是3的時候分配的情況如下圖：

圖片

先計算每個Operator實例至少分配的Key-Group個數(shù)，將不能整除的部分N個，平均分給前N個實例。最終每個Operator實例管理的Key-Groups會在GroupRange中表示，本質(zhì)是一個區(qū)間值。比如上圖是2->3擴容，那每個task的key-group的數(shù)量是：10/3≈3，也即是每個task先分3個key-group，然后把剩余的1個key-group分配給第一task。

值得注意的是：

Key-Group機制的特點就是每個具體的key(event)不關(guān)心落到具體的哪個task來處理，只關(guān)心會落到哪個Key-Group中：

• 首先 一個job運行之后，如果要復用state，不允許在修改maxParallelism。
• key 值的hash code決定落到哪個KG中，key本身不關(guān)系被哪個task處理，也就是說相同的KG在擴容前后可能被不同的task處理。

總結(jié)

State是Flink流計算的關(guān)鍵部分。Flink 中的狀態(tài)分為兩種主要類型：Keyed State 和 Operator State。Flink提供了三種狀態(tài)后端：MemoryStateBackend、FsStateBackend、RocksDBStateBackend。對于Keyed State 和 Operator State應對擴縮容時有不同的分配方式。