Flink實踐Savepoint使用範例詳解

2022-07-27 18:03:05

一、背景

什麼是 savepoint，為什麼要使用 savepoint ？

保障 flink 作業在設定迭代、flink 版本升級、藍綠部署中的資料一致性，提高容錯、降低恢復時間；

在此之前引入幾個概念：

Snapshot 狀態快照

Flink 通過狀態快照實現容錯處理

Flink 中的狀態： keyed state, operator state ..
Flink 中的狀態後端：A. 狀態資料如何存？B. 執行時存在哪裡？C. 狀態快照儲存在哪？

注1：自 1.13 版本之後，設定 Working State 和設定 Snapshot State 拆離成了兩個介面，便於讀者更易於理解；

StateBackend

CheckpointStorage

注2：一般預設使用 FsStateBackend，執行時狀態放在堆中保障效能，快照備份時資料存於 Hdfs 保障容錯性；當業務有大狀態的 flink 作業存在時，可以通過設定化的方式將使用者作業的狀態後端設定為 RocksDBSateBackend。

分散式快照

Checkpoint – a snapshot taken automatically by Flink for the purpose of being able to recover from faults. Checkpoints can be incremental, and are optimized for being restored quickly.

Alignment checkpoint

Unaligment checkpoint

未對齊的 checkpoint 確保障礙物儘快到達接收器。

適用於至少有一條緩慢移動的資料路徑的應用程式，避免對齊時間過長。然而，
會增加了額外的輸入/輸出壓力，會造成 checkpoint size 的增加，當狀態後後端 IO 有瓶頸時，不合適；

注：一般預設使用 Alignment checkpoint；當出現被壓時，一般優先採用

1. 優化邏輯 2. 增加並行能力的方式進行處理；

Checkpoint & Savepoint

Checkpoint 使 Flink 的狀態具有良好的容錯性，通過 checkpoint 機制，Flink 可以對作業的狀態和計算位置進行恢復。

Savepoint 是依據 Flink checkpointing 機制所建立的流作業執行狀態的一致映象；

Checkpoint 的主要目的是為意外失敗的作業提供恢復機制(如 tm/jm 程序掛了)。
Checkpoint 的生命週期由 Flink 管理，即 Flink 建立，管理和刪除 Checkpoint - 無需使用者互動。
Savepoint 由使用者建立，擁有和刪除。他們的用例是計劃的，手動備份和恢復。
Savepoint 應用場景，升級 Flink 版本，調整使用者邏輯，改變並行度，以及進行紅藍部署等。 Savepoint 更多地關注可移植性和對前面提到的作業更改的支援。

除去這些概念上的差異，Checkpoint 和 Savepoint 的當前實現基本上使用相同的程式碼並生成相同的格式（rocksDB 增量 checkpoint 除外，未來可能有更多類似的實現）

二、Flink on yarn 如何使用 savepoint

觸發 savepoint 保留到 hdfs, 在重新排程作業時，提供給使用者選擇即可。

關鍵點：執行 savepoint 需要指定 jobId，因此在設計資料平臺的後設資料時，需要保留 jobId 資料。

使用 YARN 觸發 Savepoint #
$ bin/flink savepoint :jobId [:targetDirectory] -yid :yarnAppId
這將觸發 ID 為 :jobId 和 YARN 應用程式 ID :yarnAppId 的作業的 Savepoint，並返回建立的 Savepoint 的路徑。
使用 Savepoint 取消作業 #
$ bin/flink cancel -s [:targetDirectory] :jobId
這將自動觸發 ID 為 :jobid 的作業的 Savepoint，並取消該作業。此外，你可以指定一個目標檔案系統目錄來儲存 Savepoint 。該目錄需要能被 JobManager(s) 和 TaskManager(s) 存取。
從 Savepoint 恢復 #
$ bin/flink run -s :savepointPath [:runArgs]
這將提交作業並指定要從中恢復的 Savepoint 。 你可以給出 Savepoint 目錄或 _metadata 檔案的路徑。
跳過無法對映的狀態恢復 #
預設情況下，resume 操作將嘗試將 Savepoint 的所有狀態對映回你要還原的程式。 如果刪除了運運算元，則可以通過 --allowNonRestoredState（short：-n）選項跳過無法對映到新程式的狀態：
$ bin/flink run -s :savepointPath -n [:runArgs]
刪除 Savepoint #
$ bin/flink savepoint -d :savepointPath
這將刪除儲存在 :savepointPath 中的 Savepoint。