一、背景

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

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

在此之前引入幾個概念：

Snapshot 狀態(tài)快照

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

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

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

StateBackend

CheckpointStorage

注2：一般默認使用 FsStateBackend，運行時狀態(tài)放在堆中保障性能，快照備份時數(shù)據(jù)存于 Hdfs 保障容錯性；當業(yè)務(wù)有大狀態(tài)的 flink 作業(yè)存在時，可以通過配置化的方式將用戶作業(yè)的狀態(tài)后端設(shè)置為 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 確保障礙物盡快到達接收器。

• 適用于至少有一條緩慢移動的數(shù)據(jù)路徑的應(yīng)用程序，避免對齊時間過長。然而，
• 會增加了額外的輸入/輸出壓力，會造成 checkpoint size 的增加，當狀態(tài)后后端 IO 有瓶頸時，不合適；

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

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

Checkpoint & Savepoint

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

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

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

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

二、Flink on yarn 如何使用 savepoint

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

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

使用 YARN 觸發(fā) Savepoint #
$ bin/flink savepoint :jobId [:targetDirectory] -yid :yarnAppId
這將觸發(fā) ID 為 :jobId 和 YARN 應(yīng)用程序 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）

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

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

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

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

端到端精確一次

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

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