簡介：在 Flink 現有的架構設計中，多流 Join 拓撲下單個 Task 失敗會導致所有 Task 重新部署，耗時可能會持續幾分鐘，導致作業的輸出斷流，這對于線上業務來說是不可接受的。針對這一痛點，字節提出單點恢復的方案。

背景

在字節跳動的實時計算場景中，我們有很多任務（數量 2k+）會直接服務于線上，其輸出時延和穩定性會直接影響線上產品的用戶體驗，這類任務通常具有如下特點：

• 流量大，并發高（最大的任務并行度超過 1w）
• 拓撲類似于多流 Join，將各個數據源做整合輸出給下游，不依賴 Checkpoint
• 沒有使用 Checkpoint 并且對短時間內的小部分數據丟失不敏感（如 0.5%），但對數據輸出的持續性要求極高

在 Flink 現有的架構設計中，多流 Join 拓撲下單個 Task 失敗會導致所有 Task 重新部署，耗時可能會持續幾分鐘，導致作業的輸出斷流，這對于線上業務來說是不可接受的。

針對這一痛點，我們提出單點恢復的方案，通過對 network 層的增強，使得在機器下線或者 Task 失敗的情況下，以短時間內故障 Task 的部分數據丟失為代價，達成以下目標：

• 作業不發生全局重啟，只有故障 Task 發生 Failover
• 非故障 Task 不受影響，正常為線上提供服務

解決思路

當初遇到這些問題的時候，我們提出的想法是說能不能在機器故障下線的時候，只讓在這臺機器上的 Tasks 進行 Failover，而這些 Tasks 的上下游 Tasks 能恰好感知到這些失敗的 Tasks，并作出對應的措施：

• 上游：將原本輸出到 Failed Tasks 的數據直接丟棄，等待 Failover 完成后再開始發送數據。
• 下游：清空 Failed Tasks 產生的不完整數據，等待 Failover 完成后再重新建立連接并接受數據

根據這些想法我們思考得出幾個比較關鍵點在于：

• 如何讓上下游感知 Task Failed ?
• 如何清空下游不完整的數據 ?
• Failover 完成后如何與上下游重新建立連接 ?

基于以上考慮我們決定基于已有的 Network 層線程模型，修改上下游對于 Task Failed 后的處理邏輯，讓非故障的 Tasks 短時間內完成對失敗 Task 的感知操作，從而使得作業持續穩定地輸出。

當前架構

注：我們的實現基于 Flink-1.9，1.11 后的網絡模型加入了 Unaligned Checkpoint 的特性，可能會有所變化。

我們先將 Flink 的上下游 Task 通信模型簡單抽象一下：

上下游 Task 感知彼此狀態的邏輯，分三種情況考慮：

• Task 因為邏輯錯誤或 OOM 等原因 Fail，Task 自身會主動釋放 network resources，給上游發送 channel close 信息，給下游發送 Exception。
• TaskManager 進程被 Yarn Kill，TCP 連接會被操作系統正常關閉，上游 Netty Server 和下游 Netty Client 可以感知到連接狀態變化。
• 機器斷電宕機，這個情況下操作系統不會正確關閉 TCP 連接，所以 Netty 的 Server 和 Client 可能互相感知不到，這個時候我們在 deploy 新的 Task 后需要做一些強制更新的處理。

可以看到，在大部分情況下，Task 是可以直接感知到上下游 Task 的狀態變化。了解了基礎的通信模型之后，我們可以按照之前的解決思路繼續深入一下，分別在上游發送端和下游接收端可以做什么樣改進來實現單點恢復。

優化方案

根據我們的解決思路，我們來繪制一下單個 Task 掛了之后，整個 Job 的通信流程：

Map(1) 失敗之后：

將 Map(1) 失敗的信息通知 Source(1) 、Sink(1) 和 JobManager。

JobManager 開始申請新的資源準備 Failover，同時上游 Source(1) 和下游 Sink(1) 切斷和 Map(1) 的數據通道，但是 Source(1) 和 Sink(1) 和其他 Task 的數據傳輸仍正常進行。

Map(1)' 被成功調度，和上游建立連接，JobManager 通知 Sink(1) 和 Map(1)' 建立連接，數據傳輸通道被恢復。

從這個流程，我們可以將優化分為三個模塊，分別為上游發送端、下游接收端和 JobManager。

上游發送端的優化

我們再細化一下上游發送端的相關細節，

(1) Netty Server 收到 Client 發送的 Partition Request 后，在對應的 Subpartition 注冊讀取數據的 SubpartitionView 和 Reader。

(2) RecordWriter 發送數據到不同的 Subpartitions，每個 Subpartition 內部維護一個 buffer 隊列，并將讀取數據的 Reader 放入到 Readers Queue 中。（Task 線程）

(3) Netty 線程讀取 Readers Queue，取出對應的 Reader 并讀取對應 Subpartition 中的 buffer 數據，發送給下游。（Netty 線程）

我們的期望是上游發送端在感知到下游 Task 失敗之后，直接將發送到對應 Task 的數據丟棄。那么我們的改動邏輯，在這個示意圖中，就是 Subpartition 通過 Netty Server 收到下游 Task Fail 的消息后，將自己設置為 Unavailable，然后 RecordWriter 在發送數據到指定 Subpartition 時，判斷是否可用，如果不可用則直接將數據丟棄。而當 Task Failover 完成后重新與上游建立連接后，再將該 Subpartition 置為 Available，則數據可以重新被消費。

發送端的改動比較簡單，得益于 Flink 內部對 Subpartition 的邏輯做了很好的抽象，并且可以很容易的通過參數來切換 Subpartition 初始化的類型，我們在這里參考 PipelinedSubpartition 的實現，根據上述的邏輯，實現了我們自己的 Subpartition 和對應的 View。

下游接收端的優化

同樣，我們來細化一下下游接收端的細節：

仔細來看，其實和上游的線程模型頗有類似之處：

(1) InputGate 初始化所有的 Channel 并通過 Netty Client 和上游 Server 建立連接。

(2) InputChannel 接收到數據后，緩存到 buffer 隊列中并將自己的引用放入到 Channels Queue 里。（Netty 線程）

(3) InputGate 通過 InputProcessor 的調用，從 Queue 里拉取 Channel 并讀取 Channel 中緩存的 buffer 數據，如果 buffer 不完整（比如只有半條 record），那么則會將不完整的 buffer 暫存到 InputProcessor 中。（Task 線程）

這里我們期望下游接收端感知到上游 Task 失敗之后，能將對應 InputChannel 的接收到的不完整的 buffer 直接清除。不完整的 buffer 存儲在 InputProcessor 中，那么我們如何讓 InputProcessor 知道哪個 Channel 出現了問題？

簡單的方案是說，我們在 InputChannel 中直接調用 InputGate 或者 InputProcessor，做 buffer 清空的操作，但是這樣引入一個問題，由于 InputChannel 收到 Error 是在 Netty 線程，而 InputProcessor 的操作是在 Task 線程，這樣跨線程的調用打破了已有的線程模型，必然會引入鎖和調用時間的不確定性，增加架構設計的復雜度，并且因為 InputProcessor 會對每一條 record 都有調用，稍有不慎就會帶來性能的下降。

我們沿用已有的線程模型，Client 感知到上游 Task 失敗的消息之后告知對應的 Channel，Channel 向自己維護的 receivedBuffers 的末尾插入一個 UnavailableEvent，并等待 InputProcessor 拉取并清空對應 Channel 的 buffer 數據。示意圖如下所示，紅色的模塊是我們新增的部分：

JobManager 重啟策略的優化

JobManager 重啟策略可以參考社區已有的 RestartIndividualStrategy，比較重要的區別是，在重新 deploy 這個失敗的 Task 后，我們需要通過 ExecutionGraph 中的拓撲信息，找到該 Task 的下游 Tasks，并通過 Rpc 調用讓下游 Tasks 和這個新的上游 Tasks 重新建立連接。

這里實現有一個難點是如果 JobManager 去 update 下游的 Channel 信息時，舊的 Channel 對應的 buffer 數據還沒有被清除怎么辦？我們這里通過新增 CachedChannelProvider 來處理這一邏輯：

如圖所示，以 Channel - 1 為例，如果 JobManager 更新 Channel 的 Rpc 請求到來時 Channel 處于不可用狀態，那么我們直接利用 Rpc 請求中攜帶的 Channel 信息來重新初始化 Channel。以 Channel - 3 為例，如果 Rpc 請求到來時 Channel 仍然可用，那么我們將 Channel 信息暫時緩存起來，等 Channel - 3 中所有數據消費完畢后，通知 CachedChannelProvider，然后再通過 CachedChannelProvider 去更新 Channel。

這里還需要特別提到一點，在字節跳動內部我們實現了預留 TaskManager 的功能，當 Task 出現 Failover 時，能夠直接使用 TaskManager 的資源，大大節約了 Failover 過程數據丟失的損耗。

實現中的關鍵點

整個解決的思路其實是比較清晰的，相信大家也比較容易理解，但是在實現中仍然有很多需要注意的地方，舉例如下：

• 上面提到 JobManager 發送的 Rpc 請求如果過早，那么會暫時緩存下來等待 Channel 數據消費完成。而此時作業的狀態是未知的，可能一直處于僵死的狀態（比如卡在了網絡 IO 或者 磁盤 IO 上），那么 Channel 中的 Unavailable Event 就無法被 InputProcessor 消費。這個時候我們通過設置一個定時器來做兜底策略，如果沒有在定時器設置的時間內完成 Channel 的重新初始化，那么該 Task 就會自動下線，走單點恢復的 Failover 流程。
• 網絡層作為 Flink 內線程模型最復雜的一個模塊，我們為了減少改動的復雜度和改動的風險，在設計上沒有新增或修改 Netty 線程和 Task 線程之間通信的模型，而是借助于已有的線程模型來實現單點恢復的功能。但在實現過程中因為給 Subpartition 和 Channel 增加了類似 isAvailable 的狀態位，所以在這些狀態的修改上需要特別注意線程可見性的處理，避免多線程讀取狀態不一致的情況發生。

收益

目前在字節跳動內部，單點恢復功能已經上線了 1000+ 作業，在機器下線、網絡抖動的情況下，下游在上游作業做 Failover 的過程幾乎沒有感知。

接下來我們以下面這個作業拓撲為例，在作業正常運行時我們手動 Kill 一個 Container，來看看不同并行度作業開啟單點恢復的效果：

我們在 1000 和 4000 并行度的作業上進行測試，每個 slot 中有 2 個 Source 和 1 個 Joiner 共 3 個 Task，手動 Kill 一個 Container 后，從故障恢復時間和斷流影響兩個維度進行收益計算：

結論：可以看到，在 4000 個 Slot 的作業里，如果不開啟單點恢復，作業整體的 Failover 時間為 81s，同時對于下游服務來說，上游服務斷流 81s，這在實時服務線上的場景中明顯是不可接受的。而開啟了單點恢復和預留資源后，Kill 1 個 Container 只會影響 4 個 Slot，且 Failover 的時間只有 5s，同時對于下游服務來說，上游服務產生的數據減少 4/4000=千分之一，持續 5s，效果是非常顯而易見的。

原文鏈接：https://developer.aliyun.com/article/774837?

版權聲明：本文內容由阿里云實名注冊用戶自發貢獻，版權歸原作者所有，阿里云開發者社區不擁有其著作權，亦不承擔相應法律責任。具體規則請查看《阿里云開發者社區用戶服務協議》和《阿里云開發者社區知識產權保護指引》。如果您發現本社區中有涉嫌抄襲的內容，填寫侵權投訴表單進行舉報，一經查實，本社區將立刻刪除涉嫌侵權內容。

總結

以上是生活随笔為你收集整理的字节跳动 Flink 单点恢复功能实践的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。