一、作業(yè)執(zhí)行容錯

Flink 的錯誤恢復(fù)機(jī)制分為多個級別，即 Execution 級別的 Failover 策略和 ExecutionGraph 級別的 Job Restart 策略。當(dāng)出現(xiàn)錯誤時，F(xiàn)link 會先嘗試觸發(fā)范圍小的錯誤恢復(fù)機(jī)制，如果仍處理不了才會升級為更大范圍的錯誤恢復(fù)機(jī)制，具體可以看下面的序列圖。

當(dāng) Task 發(fā)生錯誤，TaskManager 會通過 RPC 通知 JobManager，后者將對應(yīng) Execution 的狀態(tài)轉(zhuǎn)為 failed 并觸發(fā) Failover 策略。如果符合 Failover 策略，JobManager 會重啟 Execution，否則升級為 ExecutionGraph 的失敗。ExecutionGraph 失敗則進(jìn)入 failing 的狀態(tài)，由 Restart 策略決定其重啟（restarting 狀態(tài)）還是異常退出（failed 狀態(tài)）。

1.1、Task Failover策略

Task Failover策略目前有三個，分別是：RestartAll、RestartIndividualStrategy 和 RestartPipelinedRegionStrategy。

RestartAll: 重啟全部 Task，是恢復(fù)作業(yè)一致性的最安全策略，會在其他 Failover 策略失敗時作為保底策略使用。目前是默認(rèn)的 Task Failover 策略。

RestartPipelinedRegionStrategy: 重啟錯誤 Task 所在 Region 的全部 Task。Task Region 是由 Task 的數(shù)據(jù)傳輸決定的，有數(shù)據(jù)傳輸?shù)?Task 會被放在同一個 Region，而不同 Region 之間沒有數(shù)據(jù)交換。

RestartIndividualStrategy: 恢復(fù)單個 Task。因為如果該 Task 沒有包含數(shù)據(jù)源，這會導(dǎo)致它不能重流數(shù)據(jù)而導(dǎo)致一部分?jǐn)?shù)據(jù)丟失?？紤]到至少提供準(zhǔn)確一次的投遞語義，這個策略的使用范圍比較有限，只應(yīng)用于 Task 間沒有數(shù)據(jù)傳輸?shù)淖鳂I(yè)。

1.2、Job Restart策略

如果 Task 錯誤最終觸發(fā)了 Full Restart，此時 Job Restart 策略將會控制是否需要恢復(fù)作業(yè)。Flink 提供三種 Job 具體的 Restart Strategy。

FixedDelayRestartStrategy: 允許指定次數(shù)內(nèi)的 Execution 失敗，如果超過該次數(shù)則導(dǎo)致 Job 失敗。FixedDelayRestartStrategy 重啟可以設(shè)置一定的延遲，以減少頻繁重試對外部系統(tǒng)帶來的負(fù)載和不必要的錯誤日志。

FailureRateRestartStrategy: 允許在指定時間窗口內(nèi)的指定次數(shù)內(nèi)的 Execution 失敗，如果超過這個頻率則導(dǎo)致 Job 失敗。同樣地，F(xiàn)ailureRateRestartStrategy 也可以設(shè)置一定的重啟延遲。

NoRestartStrategy: 在 Execution 失敗時直接讓 Job 失敗。

二、守護(hù)進(jìn)程容錯

Flink on YARN 的部署模式，關(guān)鍵的守護(hù)進(jìn)程有 JobManager 和 TaskManager 兩個，其中JobManager的主要職責(zé)協(xié)調(diào)資源和管理作業(yè)的執(zhí)行分別為ResourceManager 和 JobMaster 兩個守護(hù)線程承擔(dān)，三者之間的關(guān)系如下圖所示。

2.1、TaskManager 的容錯

如果 ResouceManager 通過心跳超時檢測到或者通過集群管理器的通知了解到 TaskManager 故障，它會通知對應(yīng)的 JobMaster 并啟動一個新的 TaskManager 以做代替。注意 ResouceManager 并不關(guān)心 Flink 作業(yè)的情況，這是 JobMaster 的職責(zé)去管理 Flink 作業(yè)要做何種反應(yīng)。

如果 JobMaster 通過 ResouceManager 的通知了解到或者通過心跳超時檢測到 TaskManager 故障，它首先會從自己的 slot pool 中移除該 TaskManager，并將該 TaskManager 上運(yùn)行的所有 Tasks 標(biāo)記為失敗，從而觸發(fā) Flink 作業(yè)執(zhí)行的容錯機(jī)制以恢復(fù)作業(yè)。

TaskManager 的狀態(tài)已經(jīng)寫入 checkpoint 并會在重啟后自動恢復(fù)，因此不會造成數(shù)據(jù)不一致的問題。

2.2、ResourceManager 的容錯

如果TaskManager通過心跳超時檢測到 ResourceManager 故障，或者收到 zookeeper 的關(guān)于ResourceManager失去leadership通知，TaskManager會尋找新的 leader，ResourceManager 并將自己重啟注冊到其上，期間并不會中斷 Task的執(zhí)行。

如果JobMaster通過心跳超時檢測到ResourceManager故障，或者收到 zookeeper 的關(guān)于 ResourceManager 失去 leadership 通知，JobMaster 同樣會等待新的 ResourceManager 變成 leader，然后重新請求所有的TaskManager?？紤]到 TaskManager 也可能成功恢復(fù)，這樣的話 JobMaster 新請求的 TaskManager 會在空閑一段時間后被釋放。

ResourceManager上保持了很多狀態(tài)信息，包括活躍的 container、可用的 TaskManager、TaskManager 和 JobMaster 的映射關(guān)系等等信息，不過這些信息并不是 ground truth，可以從與 JobMaster 及 TaskManager 的狀態(tài)同步中再重新獲得，所以這些信息并不需要持久化。

2.3、JobMaster 的容錯

如果 TaskManager 通過心跳超時檢測到 JobMaster 故障，或者收到 zookeeper 的關(guān)于 JobMaster 失去 leadership 通知，TaskManager 會觸發(fā)自己的錯誤恢復(fù)，然后等待新的 JobMaster。如果新的 JobMaster 在一定時間后仍未出現(xiàn)，TaskManager 會將其 slot 標(biāo)記為空閑并告知 ResourceManager。

如果 ResourceManager 通過心跳超時檢測到 JobMaster 故障，或者收到 zookeeper 的關(guān)于 JobMaster 失去 leadership 通知，ResourceManager 會將其告知 TaskManager，其他不作處理。

JobMaster 保存了很多對作業(yè)執(zhí)行至關(guān)重要的狀態(tài)，其中 JobGraph 和用戶代碼會重新從 HDFS 等持久化存儲中獲取，checkpoint 信息會從 zookeeper 獲得，Task 的執(zhí)行信息可以不恢復(fù)因為整個作業(yè)會重新調(diào)度，而持有的 slot 則從 ResourceManager 的 TaskManager 的同步信息中恢復(fù)。

2.4、并發(fā)故障

Flink on YARN 部署模式下，因為 JobMaster 和 ResourceManager 都在 JobManager 進(jìn)程內(nèi)，如果JobManager 進(jìn)程出問題，通常是 JobMaster 和 ResourceManager 并發(fā)故障，那么 TaskManager 會按以下步驟處理：

• 按照普通的 JobMaster 故障處理。
• 在一段時間內(nèi)不斷嘗試將 slot 提供給新的 JobMaster。
• 不斷嘗試將自己注冊到 ResourceManager 上。

值得注意的是，新 JobManager 的拉起是依靠 YARN 的 Application attempt 重試機(jī)制來自動完成的，而根據(jù) Flink 配置的 YARN Application：keep-containers-across-application-attempts行為，TaskManager 不會被清理，因此可以重新注冊到新啟動的 Flink ResourceManager 和 JobMaster 中。

三、總結(jié)

Flink 容錯機(jī)制確保了 Flink 的可靠性和持久性，具體來說它包括作業(yè)執(zhí)行的容錯和守護(hù)進(jìn)程的容錯兩個方面。在作業(yè)執(zhí)行容錯方面，F(xiàn)link 提供 Task 級別的 Failover 策略和 Job 級別的 Restart 策略來進(jìn)行故障情況下的自動重試。在守護(hù)進(jìn)程的容錯方面，在on YARN 模式下，F(xiàn)link 通過內(nèi)部組件的心跳和 YARN 的監(jiān)控進(jìn)行故障檢測。TaskManager 的故障會通過申請新的 TaskManager 并重啟 Task 或 Job 來恢復(fù)，JobManager 的故障會通過集群管理器的自動拉起新 JobManager 和 TaskManager 的重新注冊到新 leader JobManager 來恢復(fù)。

以上就是淺談Flink容錯機(jī)制之作業(yè)執(zhí)行和守護(hù)進(jìn)程的詳細(xì)內(nèi)容，更多關(guān)于Flink容錯機(jī)制 作業(yè)執(zhí)行和守護(hù)進(jìn)程的資料請關(guān)注腳本之家其它相關(guān)文章！