部落格

2023/12/18

使用 Beam 和 Flink 建構可擴展、自我管理的串流基礎架構：解決自動擴展的挑戰 - 第二部分

Talat Uyarer

使用 Flink 建構可擴展、自我管理的串流基礎架構：解決自動擴展的挑戰 - 第二部分

歡迎來到我們深入探討如何在 Kubernetes 上建構和管理 Apache Beam Flink 服務的系列文章的第二部分。在這一部分中，我們將仔細研究我們在實施自動擴展時遇到的障礙。這些挑戰不僅僅是阻礙，它們也是我們創新和增強系統的機會。讓我們分解這些問題，了解它們的背景，並探索我們開發的解決方案。

挑戰：在我們目前的設定中，我們使用 Apache Flink 來處理資料串流。但是，我們遇到了一個令人困惑的問題：我們的 Flink 工作沒有顯示 Apache Beam 的積壓指標。這些指標對於了解我們的資料管線的狀態和效能至關重要。

我們的發現：有趣的是，我們注意到指標實際上是在 KafkaIO 中產生的，這是我們處理 Kafka 串流的資料管線的一部分。但是，當我們嘗試透過 Apache Flink 指標系統監控這些指標時，我們找不到它們。我們懷疑 Apache Beam 和 Apache Flink 之間的整合（或「連結」）可能存在問題。

深入研究：經過仔細檢查，我們發現指標應該在資料串流處理的「檢查點」階段發出。在這個關鍵步驟中，系統會擷取串流狀態的快照，而指標通常是為無邊界來源產生的指標。無邊界來源是指持續串流資料的來源，例如 Kafka。

一個潛在的解決方案：我們認為問題的根源在於在檢查點階段設定指標上下文的方式。連線中斷似乎阻止了 Beam 指標在 Flink 指標系統中被正確捕獲。我們針對此問題提出了一個修復方案，您可以在我們的 GitHub pull request 上查看並參與貢獻：Apache Beam PR #29793。

Apache Flink Beam Backlog Metrics

克服自動擴展 Beam 工作中檢查點大小縮減的挑戰

在本節中，我們將討論在自動擴展 Apache Beam 工作中縮減檢查點大小的策略，重點關注 Apache Flink 中的有效檢查點建立，並最佳化套件大小和 PipelineOptions，以管理頻繁的檢查點逾時和大規模工作需求。

在串流處理中，維護狀態一致性和容錯能力至關重要。Apache Flink 通過稱為檢查點建立的過程實現了這一點。檢查點建立會定期擷取工作的運算子狀態，並將其儲存在穩定的儲存位置，例如 Google Cloud Storage 或 AWS S3。具體來說，Flink 每十秒檢查一次工作，並允許最多一分鐘完成此過程。此過程對於確保在發生故障時，工作可以從最後一個檢查點恢復至關重要，從而提供恰好一次語義和容錯能力。

套件在 Apache Beam 中的作用

Apache Beam 引入了套件的概念。套件本質上是一組一起處理的元素。此步驟透過減少單獨處理每個元素的負擔來提高處理效率和吞吐量。如需更多資訊，請參閱套件和持久性。在 Flink 執行器預設配置中，套件的預設大小為 1000 個元素，逾時時間為一秒。但是，根據我們的效能測試，我們將套件大小調整為10,000 個元素，逾時時間為 10 秒。

挑戰：頻繁的檢查點逾時

當我們設定每 10 秒建立一次檢查點時，我們面臨頻繁的檢查點逾時，通常超過 1 分鐘。這是由於檢查點的大小很大造成的。

解決方案：管理檢查點大小

在 Apache Beam Flink 工作中，finishBundleBeforeCheckpointing 選項起著關鍵作用。啟用後，它可確保在啟動檢查點之前完全處理所有套件。這會導致檢查點僅包含套件完成後的狀態，從而顯著縮減檢查點大小。最初，我們的每個管線的檢查點約為 2 MB。透過此變更，它們穩定降至 150 KB。

解決大型工作中檢查點的大小問題

儘管減少了檢查點大小，但每十秒建立一次 150 KB 的檢查點仍然很大，尤其是在執行多個管線的工作中。例如，在單個工作中有 100 個管線的情況下，此大小會膨脹到每 10 秒間隔 15 MB。

進一步最佳化：透過 PipelineOptions 縮減檢查點大小

我們發現由於特定問題 (BEAM-8577)，我們的 Flink 執行器在每個檢查點中都包含我們的大型 PipelineOptions 物件。我們透過從 PipelineOptions 中移除不必要的應用程式相關選項來解決此問題，從而進一步將每個管線的檢查點大小縮減到更容易管理的 10 KB。

Kafka 讀取器等待時間：解決 Beam 工作中的自動擴展挑戰

了解未對齊的檢查點建立

在我們的系統中，我們使用未對齊的檢查點建立來加速檢查點建立過程，這對於確保分散式系統中的資料一致性至關重要。但是，當我們啟動 finishBundleBeforeCheckpointing 功能時，我們開始面臨檢查點逾時問題和檢查點建立步驟的延遲。Apache Beam 利用 Apache Flink 的舊版來源實作來處理無邊界來源。在 Flink 中，工作分為兩種：來源工作和非來源工作。

  • 來源工作：從外部系統提取資料到 Flink 工作中
  • 非來源工作：處理傳入資料

Apache Flink Task Types

在標準配置中，非來源工作會在提取資料之前檢查是否有可用的緩衝區。如果來源工作不執行此檢查，它們在將資料寫入輸出緩衝區時可能會遇到檢查點延遲。這種延遲會影響未對齊檢查點的效率，只有當輸出緩衝區可用時，舊版來源工作才會識別這些檢查點。

使用 Beam 中的 UnboundedSourceWrapper 解決挑戰

為了解決這個問題，Apache Flink 引入了一種以提取模式運作的新來源實作。在此模式下，工作會在提取資料之前檢查是否有空閒緩衝區，這與非來源工作的方法一致。

但是，Apache Beam 的 Flink 執行器仍然使用的舊版來源則以推送模式運作。它會立即將資料傳送到下游工作。當緩衝區已滿時，此設定可能會造成瓶頸，導致偵測未對齊檢查點界限時發生延遲。

我們的解決方案

儘管已遭棄用，但 Apache Beam 的 Flink 執行器仍然使用舊版來源實作。為了解決其問題，我們實施了我們的修改和 FLINK-26759 中建議的快速解決方法。這些增強功能在我們的Pull Request 中有詳細說明。您也可以在 Flink 未對齊檢查點部落格文章中找到有關未對齊檢查點問題的更多資訊。

Apache Flink UI Checkpoint History

解決高流量情境中的讀取速度緩慢問題

在我們使用 Apache Beam 和 Flink 執行器的過程中，我們遇到了一個與 Antonio Si 在 Apache Beam 中偵錯消費者延遲的方式 一文中記錄的類似挑戰，該文章由 Antonio Si 在 Intuit 的經驗中撰寫。他們在即時資料處理管線中，特別是在訊息流量高的主題中，Kafka 消費者延遲時間不斷增加。此問題的根源在於 Apache Beam 透過 UnboundedSourceWrapper 和 KafkaUnboundedReader 處理 Kafka 分區的方式。具體來說，對於流量較低的主題，處理執行緒會不必要地暫停，從而延遲流量高的主題的處理。我們在我們的系統中面臨類似的情況，即高流量和低流量主題之間處理速度的不平衡導致效率低下。

UnboundedSourceWrapper Design

為了解決這個問題，我們開發了一種創新的解決方案：KafkaIO 中的自適應逾時策略。此策略會根據每個主題的流量動態調整逾時時間。對於低流量主題，它會縮短逾時時間，以防止不必要的延遲。對於高流量主題，它會延長逾時時間，以提供更多的處理機會。此方法在我們最近的 pull request 中有詳細說明。

Beam 工作自動擴展中不平衡的分區分佈

此系統的核心是自適應排程器，這是一個專為快速資源分配而設計的元件。它會根據運算插槽的可用性，智慧地調整工作執行的平行工作數量（平行度）。這些插槽就像單獨的工作站，每個工作站都可以處理工作的某些部分。

但是，我們遇到了一個問題。我們的工作包含多個獨立的管線，每個管線都需要自己的資源集。最初，系統傾向於將更多的工作分配給前幾個工作人員，從而使他們超負荷工作，而其他工作人員則處於未充分利用的狀態。此問題的原因是 Flink 分配工作的方式，它會優先選擇每個管線的前幾個工作人員。

Flink split assignment on slots

為了解決這個問題，我們為 Flink 的 SlotSharingSlotAllocator（一個負責工作分配的元件）開發了一個自訂修補程式。此修補程式可確保在所有可用的工作人員之間更均衡地分配工作負載，從而提高效率並防止瓶頸。透過此改進，每個工作人員都可以公平地獲得工作份額，從而更好地利用資源並使我們的 Beam 工作更順暢地運作。

挑戰

在使用 Apache Flink 進行資料處理的世界中，一個常見的任務是管理和更新資料處理任務。這些任務可以是具狀態的，能夠記住過去的資料，也可以是無狀態的，不需記住過去的資料。

過去，當我們需要更新或刪除由 Kubernetes Operator 管理的 Flink 任務時，系統會使用儲存點 (savepoint) 或檢查點 (checkpoint) 來儲存任務的目前狀態。然而，有一個關鍵步驟遺漏了：系統沒有停止任務處理新資料（這就是我們所說的「排空」任務）。這個疏忽可能導致兩個主要問題：

  1. 對於具狀態的任務：可能出現資料不一致的情況，因為任務可能會處理儲存點中未考慮到的新資料。
  2. 對於無狀態的任務：可能出現資料重複的情況，因為任務可能會重新處理已經處理過的資料。

解決方案：排空 (drain) 功能

這就是為什麼需要 FLINK-32700 中引用的更新。此更新引入了排空功能。可以把它想像成告訴任務：「完成你目前正在處理的內容，但不要接收任何新的內容。」以下是它的運作方式：

  1. 停止新資料：任務停止讀取新的輸入。
  2. 標記來源：任務使用無限大的水位線 (watermark) 標記來源。將此水位線視為一個標記，告訴系統沒有新的資料需要處理。
  3. 在管線中傳播：然後，此標記會傳遞到任務的處理管線中，確保任務的每個部分都知道不需要預期任何新的資料。

這個看似微小的改變卻有很大的影響。它可以確保在更新或刪除任務時，它所處理的資料保持一致和準確。這對於任何資料處理任務都至關重要，因為它可以維護資料的完整性和可靠性。此外，如果排空失敗，您可以取消任務而無需儲存點，這為整個過程增加了一層彈性和安全性。

結論

當我們總結 Kubernetes 上建構和管理 Apache Beam Flink 服務系列的第二部分時，很明顯，實施自動擴展的過程既具有挑戰性又富有啟發性。我們遇到的障礙，從理解 Flink Runner 環境中的 Apache Beam 待辦事項指標，到解決高流量情境中的讀取速度緩慢問題，都促使我們開發創新的解決方案並加深對串流基礎架構的理解。

我們對檢查點 (checkpointing)、Kafka Reader 等待時間和不平衡分割區分配的深入探討，揭示了自動擴展 Beam 任務的複雜性。這些挑戰促使我們制定了諸如 KafkaIO 中的自適應逾時和 Flink SlotSharingSlotAllocator 中的平衡工作負載分配等策略。此外，在 Kubernetes Operator 中引入對 Flink 的排空支援，標誌著有效管理具狀態和無狀態任務的一個重大進步。

這段旅程不僅提高了我們系統的穩健性和效率，還為更廣泛的 Apache Beam 和 Flink 社群提供了寶貴的見解。我們希望我們的經驗和解決方案可以幫助其他人在他們的專案中面臨類似的挑戰。

請繼續關注我們的下一篇博文，我們將深入探討 Apache Beam 中自動擴展的細節。我們將分解概念、策略和最佳實務，以有效地擴展您的 Beam 任務。感謝您關注我們的系列，我們期待與您分享更多我們的旅程和學習心得。

鳴謝

這是一項巨大的努力，旨在建立新的基礎架構，並將大型客戶基礎應用程式從雲端供應商管理的串流基礎架構遷移到大規模的自我管理、基於 Flink 的基礎架構。感謝 Palo Alto Networks CDL 串流團隊協助實現這一切：Kishore Pola、Andrew Park、Hemant Kumar、Manan Mangal、Helen Jiang、Mandy Wang、Praveen Kumar Pasupuleti、JM Teo、Rishabh Kedia、Talat Uyarer、Naitk Dani 和 David He。

探索更多

加入我們的 社群，分享您的經驗，或在 GitHub 上為我們正在進行的專案做出貢獻。您的回饋非常寶貴。如果您對此系列有任何意見或問題，請隨時透過 使用者郵件列表 與我們聯絡。

請與我們保持聯繫，以獲取有關 Apache Beam、Flink 和 Kubernetes 的更多更新和見解。