淺談Flink容錯機制之作業(yè)執行和守護進(jìn)程

目錄

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

Flink 的錯誤恢復機制分為多個(gè)級別，即 Execution 級別的 Failover 策略和 ExecutionGraph 級別的 Job Restart 策略。當出現錯誤時(shí)，Flink 會(huì )先嘗試觸發(fā)范圍小的錯誤恢復機制，如果仍處理不了才會(huì )升級為更大范圍的錯誤恢復機制，具體可以看下面的序列圖。

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

1.1、Task Failover策略

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

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

RestartPipelinedRegionStrategy: 重啟錯誤 Task 所在 Region 的全部 Task。Task Region 是由 Task 的數據傳輸決定的，有數據傳輸的 Task 會(huì )被放在同一個(gè) Region，而不同 Region 之間沒(méi)有數據交換。

RestartIndividualStrategy: 恢復單個(gè) Task。因為如果該 Task 沒(méi)有包含數據源，這會(huì )導致它不能重流數據而導致一部分數據丟失?？紤]到至少提供準確一次的投遞語(yǔ)義，這個(gè)策略的使用范圍比較有限，只應用于 Task 間沒(méi)有數據傳輸的作業(yè)。

1.2、Job Restart策略

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

FixedDelayRestartStrategy: 允許指定次數內的 Execution 失敗，如果超過(guò)該次數則導致 Job 失敗。FixedDelayRestartStrategy 重啟可以設置一定的延遲，以減少頻繁重試對外部系統帶來(lái)的負載和不必要的錯誤日志。

FailureRateRestartStrategy: 允許在指定時(shí)間窗口內的指定次數內的 Execution 失敗，如果超過(guò)這個(gè)頻率則導致 Job 失敗。同樣地，FailureRateRestartStrategy 也可以設置一定的重啟延遲。

NoRestartStrategy: 在 Execution 失敗時(shí)直接讓 Job 失敗。

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

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

2.1、TaskManager 的容錯

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

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

TaskManager 的狀態(tài)已經(jīng)寫(xiě)入 checkpoint 并會(huì )在重啟后自動(dòng)恢復，因此不會(huì )造成數據不一致的問(wèn)題。

2.2、ResourceManager 的容錯

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

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

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

2.3、JobMaster 的容錯

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

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

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

2.4、并發(fā)故障

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

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

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

三、總結

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

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