文章目錄

• 狀態管理
• 算子狀態
• 鍵值分區狀態
• 狀態后端（State Backends）
• 有狀態算子的擴縮容

---

狀態管理

通常意義上，函數里所有需要任務去維護并用來計算結果的數據都屬于任務的狀態，可以把狀態想象成任務的業務邏輯所需要訪問的本地或實例變量。

任務和狀態之間的經典交互過程

如上圖，任務首先會接受一些輸入數據。在處理這些數據的過程中，任務對其狀態進行讀取或更新，并根據狀態的輸入數據計算結果。我們以一個持續計算接收到多少條記錄的簡單任務為例。當任務收到一個新的記錄后，首先會訪問狀態獲取當前統計的記錄數目，然后把數目增加并更新狀態，最后將更新后的狀態數目發送出去。

Flink會負責進行狀態的管理，包括狀態一致性、故障處理以及高效存取相關的問題都由Flink負責搞定，這樣開發人員就可以專注于自己的應用邏輯。

在Flink中，狀態都是和特定operator（算子）相關聯，為了讓Flink的Runtime（運行）層知道算子有哪些狀態，算子需要自己對其進行注冊。根據作用域的不同，狀態可以分為以下兩類

• operator state（算子狀態）
• keyed state（鍵值分區狀態）

算子狀態

算子狀態的作用域是某個算子任務，這意味著所有在同一個并行任務之內的記錄都能訪問到相同的狀態==（每一個并行的子任務都共享一個狀態）。算子狀態不能通過其他任務訪問，無論該任務是否來自相同算子（相同算子的不同任務之間也不能訪問）==。

帶有算子狀態的任務

Flink為算子狀態提供了三種數據結構

列表狀態（list state）：將狀態表示為一組數據的列表。（每一個并行的子任務共享一個狀態）

聯合列表狀態（union list state）：同樣將狀態表示為數據的列表，但在進行故障恢復或者從某個保存點（savepoint）啟動應用的時候，狀態恢復的方式和普通的列表狀態有所不同。（把之前的每一個狀態廣播到對應的每一個算子中）

廣播狀態（broadcast state）：專門為那些需要保證算子的每個任務狀態都相同的場景而設計。（把同一個狀態廣播給所有算子子任務）

鍵值分區狀態

鍵值分區狀態會按照算子輸入記錄所定義的鍵值來進行維護或訪問。Flink為每個鍵值都維護了一個狀態實例，該實例總是位于那個處理對應鍵值記錄的算子任務上。當任務在處理一個記錄時，會自動把狀態的訪問范圍限制為當前記錄的鍵值，因此所有鍵值相同的記錄都能訪問到一樣的狀態。

帶有鍵值分區狀態的任務

Flink為鍵值分區狀態提供以下幾種數據結構

單值狀態（value state）：每個鍵對應存儲一個任意類型的值。

列表狀態（list state）：每個鍵對應存儲一個值的列表。

映射狀態（map state）：每個鍵對應存儲一個鍵值映射。

聚合狀態（Reducing state & Aggregating State） ：每個鍵對應存儲一個用于聚合操作的列表

狀態后端（State Backends）

有狀態算子的任務通常會對每一條到來的記錄讀寫狀態，因此高效的狀態訪問對于記錄處理的低延遲而言至關重要。為了保證快速訪問狀態，每個并行任務都會把狀態維護在本地。至于狀態具體的存儲、訪問和維護，則是由一個名為狀態后端的**可拔插（pluggable）**組件來決定。狀態后端主要負責兩件事情：本地狀態管理和將狀態以檢查點的形式寫入遠程存儲。

目前，Flink提供了三種狀態后端，狀態后端的選擇會影響有狀態應用的魯棒性及性能。

MemoryStateBackend

• MemoryStateBackend將狀態以常規對象的方式存儲在TaskManager進程的JVM堆，并在生成Checkpoints時會將狀態發送至JobManager并保存到它的堆內存中。
• 如果狀態過大，則可能導致JVM上的任務由于OutOfMemoryError而終止，并且可能由于堆中放置了過多常駐內存的對象而引發垃圾回收停頓問題。
• 由于內存具有易失性，所以一旦JobManager出現故障就會導致狀態丟失，因此MemoryStateBackend通常用于開發和調試。
• 內存訪問速度快，延遲低，但容錯性也低。

FsStateBackend

• 與MemoryStateBackend一樣將本地狀態存儲在TaskManager進程的JVM堆里，不同的是將Checkpoints存到了遠程持久化文件系統（FileSystem）中。
• 受到TaskManager內存大小的限制，并且也可能導致垃圾回收停頓問題。
• FsStateBackend既讓本地訪問享有內存的速度，又可以支持故障容錯。

RocksDBStateBackend

• RocksDBStateBackend會將全部狀態序列化后存到本地RocksDB實例中

• 由于磁盤I/O以及序列化/反序列化對象的性能開銷，相較于內存中維護狀態而言， 讀寫性能會偏低。

• RocksDB的支持并不直接包含在Flink中，需要額外引入依賴

  <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-statebackend-rocksdb_2.12</artifactId> <version>1.12.1</version> </dependency>

有狀態算子的擴縮容

流式應用的一項基本需求是根據輸入數據到達速率的變化調整算子的并行度。對于無狀態的算子擴縮容很容易，但是對于有狀態算子來說，這就變的復雜了很多。因為我們需要把狀態重新分組，分配到與之前數量不等的并行任務上。

針對不同類型狀態的算子，Flink提供了四種擴縮容模式

鍵值分區狀態

算子列表狀態

算子聯合列表狀態

算子廣播狀態

鍵值分區狀態

帶有鍵值分區狀態的算子在擴縮容時會根據新的任務數量對鍵值重新分區，但為了降低狀態在不同任務之間遷移的必要成本，Flink不會對單獨的鍵值實施再分配，而是會把所有鍵值分為不同的鍵值組（Key group）。每個鍵值組都包含了部分鍵值，Flink以此為單位把鍵值分配給不同任務。

算子擴縮容時鍵值分區狀態的調整

算子列表狀態

帶有算子列表狀態的算子在擴縮容時會對列表中的條目進行重新分配。理論上，所有并行算子任務的列表條目會被統一收集起來，隨后均勻分配到更少或更多的任務之上。如果列表條目的數量小于算子新設置的并行度，部分任務在啟動時的狀態就可能為空。

算子擴縮容時算子列表狀態的調整

算子聯合列表狀態

帶有算子聯合列表狀態的算子會在擴縮容時把狀態列表的全部條目廣播到全部任務上，隨后由任務自己決定哪些條目應該保留，哪些應該丟棄。

算子擴縮容時算子聯合列表狀態的調整

算子廣播狀態

帶有算子廣播狀態的算子在擴縮容時會把狀態拷貝到全部新任務上，這樣做的原因是廣播狀態能確保所有任務的狀態相同。在縮容的情況下，由于狀態經過復制不會丟失，我們可以簡單的停掉多出的任務。

算子擴縮容時算子廣播狀態的調整

總結

以上是生活随笔為你收集整理的Flink 状态管理：算子状态、键值分区状态、状态后端、有状态算子的扩缩容的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。