Apache Flink 進階（三）：Checkpoint 原理解析與應用實踐

簡介： 大家好，今天我將跟大家分享一下 Flink 里面的 Checkpoint，共分為四個部分。

Checkpoint 是從 source 觸發到下游所有節點完成的一次全局操作。下圖可以有一個對 Checkpoint 的直觀感受，紅框里面可以看到一共觸發了 569K 次 Checkpoint，然后全部都成功完成，沒有 fail 的。

image.png

state 其實就是 Checkpoint 所做的主要持久化備份的主要數據，看下圖的具體數據統計，其 state 也就 9kb 大小

image.png

我們接下來看什么是 state。先看一個非常經典的 word count 代碼，這段代碼會去監控本地的 9000 端口的數據并對網絡端口輸入進行詞頻統計，我們本地行動 netcat，然后在終端輸入 hello world，執行程序會輸出什么？

image.png

答案很明顯，(hello, 1) 和 (word,1)

那么問題來了，如果再次在終端輸入 hello world，程序會輸入什么？

答案其實也很明顯，(hello, 2) 和 (world, 2)。為什么 Flink 知道之前已經處理過一次 hello world，這就是 state 發揮作用了，這里是被稱為 keyed state 存儲了之前需要統計的數據，所以幫助 Flink 知道 hello 和 world 分別出現過一次。

回顧一下剛才這段 word count 代碼。keyby 接口的調用會創建 keyed stream 對 key 進行劃分，這是使用 keyed state 的前提。在此之后，sum 方法會調用內置的 StreamGroupedReduce 實現。

image.png

又稱為 non-keyed state，每一個 operator state 都僅與一個 operator 的實例綁定。

常見的 operator state 是 source state，例如記錄當前 source 的 offset

再看一段使用 operator state 的 word count 代碼：

image.png

這里的fromElements會調用FromElementsFunction的類，其中就使用了類型為 list state 的 operator state。根據 state 類型做一個分類如下圖：

image.png

Managed State：由 Flink 管理的 state，剛才舉例的所有 state 均是 managed state

Raw State：Flink 僅提供 stream 可以進行存儲數據，對 Flink 而言 raw state 只是一些 bytes

在實際生產中，都只推薦使用 managed state，本文將圍繞該話題進行討論。

下圖就前文 word count 的 sum 所使用的StreamGroupedReduce類為例講解了如何在代碼中使用 keyed state：

image.png

下圖則對 word count 示例中的FromElementsFunction類進行詳解并分享如何在代碼中使用 operator state：

image.png

Statebackend 的分類

下圖闡釋了目前 Flink 內置的三類 state backend，其中MemoryStateBackend和FsStateBackend在運行時都是存儲在 java heap 中的，只有在執行 Checkpoint 時，FsStateBackend才會將數據以文件格式持久化到遠程存儲上。而RocksDBStateBackend則借用了 RocksDB（內存磁盤混合的 LSM DB）對 state 進行存儲。

image.png

對于HeapKeyedStateBackend，有兩種實現：

支持異步 Checkpoint（默認）：存儲格式 CopyOnWriteStateMap

僅支持同步 Checkpoint：存儲格式 NestedStateMap

特別在 MemoryStateBackend 內使用HeapKeyedStateBackend時，Checkpoint 序列化數據階段默認有最大 5 MB數據的限制

對于RocksDBKeyedStateBackend，每個 state 都存儲在一個單獨的 column family 內，其中 keyGroup，Key 和 Namespace 進行序列化存儲在 DB 作為 key。

image.png

本小節將對 Checkpoint 的執行流程逐步拆解進行講解，下圖左側是 Checkpoint Coordinator，是整個 Checkpoint 的發起者，中間是由兩個 source，一個 sink 組成的 Flink 作業，最右側的是持久化存儲，在大部分用戶場景中對應 HDFS。

第一步，Checkpoint Coordinator 向所有 source 節點 trigger Checkpoint；。

image.png

第二步，source 節點向下游廣播 barrier，這個 barrier 就是實現 Chandy-Lamport 分布式快照算法的核心，下游的 task 只有收到所有 input 的 barrier 才會執行相應的 Checkpoint。

第三步，當 task 完成 state 備份后，會將備份數據的地址（state handle）通知給 Checkpoint coordinator。

image.png

最后，當 Checkpoint coordinator 收集齊所有 task 的 state handle，就認為這一次的 Checkpoint 全局完成了，向持久化存儲中再備份一個 Checkpoint meta 文件。

image.png

為了實現 EXACTLY ONCE 語義，Flink 通過一個 input buffer 將在對齊階段收到的數據緩存起來，等對齊完成之后再進行處理。而對于 AT LEAST ONCE 語義，無需緩存收集到的數據，會對后續直接處理，所以導致 restore 時，數據可能會被多次處理。下圖是官網文檔里面就 Checkpoint align 的示意圖：

image.png

需要特別注意的是，Flink 的 Checkpoint 機制只能保證 Flink 的計算過程可以做到 EXACTLY ONCE，端到端的 EXACTLY ONCE 需要 source 和 sink 支持。

作業恢復時，二者均可以使用，主要區別如下：

image.png

Apache Flink

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。