一、背景

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

保障 flink 作業(yè)在 配置迭代、flink 版本升級、藍綠部署中的數據一致性，提高容錯、降低恢復時間；

在此之前引入幾個概念：

Snapshot 狀態(tài)快照

Flink 通過狀態(tài)快照實現容錯處理

• Flink 中的狀態(tài)： keyed state, operator state ..
• Flink 中的狀態(tài)后端：A. 狀態(tài)數據如何存？B. 運行時存在哪里？C. 狀態(tài)快照保存在哪？

注1：自 1.13 版本之后，設置 Working State 和 設置 Snapshot State 拆離成了兩個接口，便于讀者更易于理解；

StateBackend

CheckpointStorage

注2：一般默認使用 FsStateBackend，運行時狀態(tài)放在堆中保障性能，快照備份時數據存于 Hdfs 保障容錯性；當業(yè)務有大狀態(tài)的 flink 作業(yè)存在時，可以通過配置化的方式將用戶作業(yè)的狀態(tài)后端設置為 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 的增加，當狀態(tài)后后端 IO 有瓶頸時，不合適；

注：一般默認使用 Alignment checkpoint；當出現被壓時，一般優(yōu)先采用

1. 優(yōu)化邏輯 2. 增加并發(fā)能力的方式進行處理；

Checkpoint & Savepoint

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

Savepoint 是依據 Flink checkpointing 機制所創(chuàng)建的流作業(yè)執(zhí)行狀態(tài)的一致鏡像；

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

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

二、Flink on yarn 如何使用 savepoint

觸發(fā) savepoint 保留到 hdfs, 在重新調度作業(yè)時，提供給用戶選擇即可。

關鍵點：執(zhí)行 savepoint 需要指定 jobId，因此在設計數據平臺的元數據時，需要保留 jobId 數據。

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

附錄：一致性語義

確保精確一次（exactly once）

當流處理應用程序發(fā)生錯誤的時候，結果可能會產生丟失或者重復。Flink 根據你為應用程序和集群的配置，可以產生以下結果：

• Flink 不會從快照中進行恢復（at most once）
• 沒有任何丟失，但是你可能會得到重復冗余的結果（at least once）
• 沒有丟失或冗余重復（exactly once）

Flink 通過回退和重新發(fā)送 source 數據流從故障中恢復，當理想情況被描述為精確一次時，這并不意味著每個事件都將被精確一次處理。相反，這意味著 每一個事件都會影響 Flink 管理的狀態(tài)精確一次。

Barrier 只有在需要提供精確一次的語義保證時需要進行對齊（Barrier alignment）。如果不需要這種語義，可以通過配置 CheckpointingMode.AT_LEAST_ONCE 關閉 Barrier 對齊來提高性能。

端到端精確一次

為了實現端到端的精確一次，以便 sources 中的每個事件都僅精確一次對 sinks 生效，必須滿足以下條件：

• sources 必須是可重放的，并且
• sinks 必須是事務性的（或冪等的）

以上就是Flink實踐Savepoint使用示例詳解的詳細內容，更多關于Flink Savepoint使用的資料請關注腳本之家其它相關文章！