摘要：Flink 容錯機制主要有作業執行的容錯以及守護程序的容錯兩方面，前者包括 Flink runtime 的 ExecutionGraph 和Execution的容錯，後者則包括 JobManager 和 TaskManager 的容錯。
本文分享自華為雲社群《Flink容錯機制》，原文作者：yangxiao_mrs 。
Flink 容錯機制主要有作業執行的容錯以及守護程序的容錯兩方面，前者包括 Flink runtime 的 ExecutionGraph 和Execution的容錯，後者則包括 JobManager 和 TaskManager 的容錯。
一、作業執行容錯
Flink 的錯誤恢複機制分為多個級别，即 Execution 級别的 Failover 政策和 ExecutionGraph 級别的 Job Restart 政策。當出現錯誤時，Flink 會先嘗試觸發範圍小的錯誤恢複機制，如果仍處理不了才會更新為更大範圍的錯誤恢複機制，具體可以看下面的序列圖。
當 Task 發生錯誤，TaskManager 會通過 RPC 通知 JobManager，後者将對應 Execution 的狀态轉為 failed 并觸發 Failover 政策。如果符合 Failover 政策，JobManager 會重新開機 Execution，否則更新為 ExecutionGraph 的失敗。ExecutionGraph 失敗則進入 failing 的狀态，由 Restart 政策決定其重新開機（restarting 狀态）還是異常退出（failed 狀态）。
1.1 Task Failover政策
Task Failover政策目前有三個，分别是：RestartAll、RestartIndividualStrategy 和 RestartPipelinedRegionStrategy。
RestartAll: 重新開機全部 Task，是恢複作業一緻性的最安全政策，會在其他 Failover 政策失敗時作為保底政策使用。目前是預設的 Task Failover 政策。
RestartPipelinedRegionStrategy: 重新開機錯誤 Task 所在 Region 的全部 Task。Task Region 是由 Task 的資料傳輸決定的，有資料傳輸的 Task 會被放在同一個 Region，而不同 Region 之間沒有資料交換。
RestartIndividualStrategy: 恢複單個 Task。因為如果該 Task 沒有包含資料源，這會導緻它不能重流資料而導緻一部分資料丢失。考慮到至少提供準确一次的投遞語義，這個政策的使用範圍比較有限，隻應用于 Task 間沒有資料傳輸的作業。
1.2 Job Restart政策
如果 Task 錯誤最終觸發了 Full Restart，此時 Job Restart 政策将會控制是否需要恢複作業。Flink 提供三種 Job 具體的 Restart Strategy。
FixedDelayRestartStrategy: 允許指定次數内的 Execution 失敗，如果超過該次數則導緻 Job 失敗。FixedDelayRestartStrategy 重新開機可以設定一定的延遲，以減少頻繁重試對外部系統帶來的負載和不必要的錯誤日志。
FailureRateRestartStrategy: 允許在指定時間視窗内的指定次數内的 Execution 失敗，如果超過這個頻率則導緻 Job 失敗。同樣地，FailureRateRestartStrategy 也可以設定一定的重新開機延遲。
NoRestartStrategy: 在 Execution 失敗時直接讓 Job 失敗。
二、守護程序容錯
Flink on YARN 的部署模式，關鍵的守護程序有 JobManager 和 TaskManager 兩個，其中JobManager的主要職責協調資源和管理作業的執行分别為ResourceManager 和 JobMaster 兩個守護線程承擔，三者之間的關系如下圖所示。
2.1 TaskManager 的容錯
如果 ResouceManager 通過心跳逾時檢測到或者通過叢集管理器的通知了解到 TaskManager 故障，它會通知對應的 JobMaster 并啟動一個新的 TaskManager 以做代替。注意 ResouceManager 并不關心 Flink 作業的情況，這是 JobMaster 的職責去管理 Flink 作業要做何種反應。
如果 JobMaster 通過 ResouceManager 的通知了解到或者通過心跳逾時檢測到 TaskManager 故障，它首先會從自己的 slot pool 中移除該 TaskManager，并将該 TaskManager 上運作的所有 Tasks 标記為失敗，進而觸發 Flink 作業執行的容錯機制以恢複作業。
TaskManager 的狀态已經寫入 checkpoint 并會在重新開機後自動恢複，是以不會造成資料不一緻的問題。
2.2 ResourceManager 的容錯
如果TaskManager通過心跳逾時檢測到 ResourceManager 故障，或者收到 zookeeper 的關于ResourceManager失去leadership通知，TaskManager會尋找新的 leader，ResourceManager 并将自己重新開機注冊到其上，期間并不會中斷 Task的執行。
如果JobMaster通過心跳逾時檢測到ResourceManager故障，或者收到 zookeeper 的關于 ResourceManager 失去 leadership 通知，JobMaster 同樣會等待新的 ResourceManager 變成 leader，然後重新請求所有的TaskManager。考慮到 TaskManager 也可能成功恢複，這樣的話 JobMaster 新請求的 TaskManager 會在空閑一段時間後被釋放。
ResourceManager上保持了很多狀态資訊，包括活躍的 container、可用的 TaskManager、TaskManager 和 JobMaster 的映射關系等等資訊，不過這些資訊并不是 ground truth，可以從與 JobMaster 及 TaskManager 的狀态同步中再重新獲得，是以這些資訊并不需要持久化。
2.3 JobMaster 的容錯
如果 TaskManager 通過心跳逾時檢測到 JobMaster 故障，或者收到 zookeeper 的關于 JobMaster 失去 leadership 通知，TaskManager 會觸發自己的錯誤恢複，然後等待新的 JobMaster。如果新的 JobMaster 在一定時間後仍未出現，TaskManager 會将其 slot 标記為空閑并告知 ResourceManager。
如果 ResourceManager 通過心跳逾時檢測到 JobMaster 故障，或者收到 zookeeper 的關于 JobMaster 失去 leadership 通知，ResourceManager 會将其告知 TaskManager，其他不作處理。
JobMaster 儲存了很多對作業執行至關重要的狀态，其中 JobGraph 和使用者代碼會重新從 HDFS 等持久化存儲中擷取，checkpoint 資訊會從 zookeeper 獲得，Task 的執行資訊可以不恢複因為整個作業會重新排程，而持有的 slot 則從 ResourceManager 的 TaskManager 的同步資訊中恢複。
2.4 并發故障
Flink on YARN 部署模式下，因為 JobMaster 和 ResourceManager 都在 JobManager 程序内，如果JobManager 程序出問題，通常是 JobMaster 和 ResourceManager 并發故障，那麼 TaskManager 會按以下步驟處理：
- 按照普通的 JobMaster 故障處理。
- 在一段時間内不斷嘗試将 slot 提供給新的 JobMaster。
- 不斷嘗試将自己注冊到 ResourceManager 上。
值得注意的是，新 JobManager 的拉起是依靠 YARN 的 Application attempt 重試機制來自動完成的，而根據 Flink 配置的 YARN Application： keep-containers-across-application-attempts 行為，TaskManager 不會被清理，是以可以重新注冊到新啟動的 Flink ResourceManager 和 JobMaster 中。
三、總結
Flink 容錯機制確定了 Flink 的可靠性和持久性，具體來說它包括作業執行的容錯和守護程序的容錯兩個方面。在作業執行容錯方面，Flink 提供 Task 級别的 Failover 政策和 Job 級别的 Restart 政策來進行故障情況下的自動重試。在守護程序的容錯方面，在on YARN 模式下，Flink 通過内部元件的心跳和 YARN 的監控進行故障檢測。TaskManager 的故障會通過申請新的 TaskManager 并重新開機 Task 或 Job 來恢複，JobManager 的故障會通過叢集管理器的自動拉起新 JobManager 和 TaskManager 的重新注冊到新 leader JobManager 來恢複。
點選關注，第一時間了解華為雲新鮮技術~