Dataflow 中的僅需處理一次

Dataflow 支援僅處理一次記錄。本頁面說明 Dataflow 如何實作僅須處理一次的作業，同時確保低延遲。

總覽

批次管道一律會使用僅需處理一次的處理方式。串流管道預設會使用「僅需處理一次」機制，但也可以使用「至少處理一次」機制。

「僅需處理一次」可確保處理記錄的結果，包括每個管道階段的結果。具體來說，對於從來源抵達管道的每筆記錄，或從前一階段抵達階段的每筆記錄，Dataflow 會確保下列事項：

  • 記錄已處理完畢，不會遺失。
  • 管道中的任何處理結果最多只會反映一次。

換句話說，系統至少會處理一次記錄，且只會提交一次結果。

「僅需處理一次」可確保結果準確，輸出內容不會有重複記錄。Dataflow 經過最佳化，可將延遲時間縮到最短，同時維持「僅處理一次」的語意。不過，僅需處理一次的作業仍會產生重複資料刪除費用。如果使用案例可容許重複記錄，啟用「至少一次」模式通常可以降低成本並改善延遲情況。如要進一步瞭解如何選擇「正好一次」和「至少一次」串流，請參閱「設定管道串流模式」。

延遲資料

僅處理一次可確保管道的準確度：如果管道處理記錄，Dataflow 會確保記錄反映在輸出內容中，且不會重複。

不過，在串流管道中，由於記錄可能會延遲送達，因此「只處理一次」無法保證結果完整。舉例來說，假設管道會對時間範圍執行匯總，例如 Count。採用僅處理一次的機制，可確保在時間區間內及時抵達的記錄結果準確無誤，但延遲抵達的記錄可能會遭到捨棄。

一般來說，串流管道無法保證完整性，因為理論上記錄可能會任意延遲送達。在限制情況下，您需要無限期等待才能產生結果。更實際地說，Apache Beam 可讓您設定捨棄延遲資料的門檻，以及發送匯總結果的時間。詳情請參閱 Apache Beam 說明文件中的「Watermarks and late data」。

副作用

連帶效果不一定具有「僅限一次」語意。重要事項：這包括將輸出內容寫入外部存放區，除非接收器也實作了「只執行一次」語意。

具體來說，Dataflow 無法保證每筆記錄都只會經過每個轉換一次。由於重試或工作站故障，Dataflow 可能會多次透過轉換傳送記錄，甚至在多個工作站上同時傳送。

在「僅須處理一次」作業中，Dataflow 會重複資料輸出內容。不過，如果轉換中的程式碼有副作用，這些副作用可能會多次發生。舉例來說，如果轉換會呼叫遠端服務，系統可能會針對同一筆記錄多次呼叫該服務。在某些情況下，副作用甚至可能導致資料遺失。舉例來說，假設轉換作業讀取檔案以產生輸出內容，然後立即刪除檔案，而不等待輸出內容提交。如果提交結果時發生錯誤，Dataflow 會重試轉換，但現在轉換無法讀取已刪除的檔案。

記錄

處理作業的記錄輸出內容會指出處理作業已發生，但不會指出資料是否已提交。因此，即使處理後的資料只會提交到永久儲存空間一次，記錄檔仍可能指出資料已處理多次。此外，記錄檔不一定會反映已處理及已提交的資料。記錄可能會因節流而捨棄，或因其他記錄服務問題而遺失。

僅限一次串流

本節說明 Dataflow 如何為串流工作實作僅處理一次概念，包括 Dataflow 如何管理非決定性處理、延遲資料和自訂程式碼等複雜情況。

Dataflow 串流 Shuffle

串流 Dataflow 工作會將工作範圍指派給每個工作站，在許多不同的工作站上平行執行。雖然工作指派可能會因應工作人員故障、自動調整或其他事件而隨時間變更，但在每次 GroupByKey 轉換後，具有相同鍵的所有記錄都會在同一位工作人員上處理。GroupByKey 轉換作業通常用於複合轉換作業，例如 Count、FileIO 等。為確保特定鍵的記錄最終會出現在同一個工作站上，Dataflow 工作站會使用遠端程序呼叫 (RPC)，在彼此之間隨機重組資料。

為確保在隨機播放期間不會遺失記錄，Dataflow 會使用上游備份。使用上游備份時，傳送記錄的工作站會重試 RPC，直到收到記錄已收到的正面確認為止。處理記錄的副作用會提交至下游的永久儲存空間。如果傳送記錄的工作站無法使用，Dataflow 會繼續重試 RPC，確保每筆記錄至少傳送一次。

由於這些重試作業可能會建立重複項目，因此每則訊息都會標記專屬 ID。每個接收器都會儲存已處理的所有 ID 目錄。收到記錄後，Dataflow 會在目錄中查閱記錄的 ID。如果找到 ID，表示系統已收到並提交記錄，因此會將其視為重複記錄而捨棄。為確保記錄 ID 穩定，每個步驟的輸出內容都會檢查點並儲存至儲存空間。因此，如果因重複的 RPC 呼叫而多次傳送同一則訊息，系統只會將該訊息提交至儲存空間一次。

確保低延遲

如要實現「只處理一次」的處理作業，就必須減少 I/O，特別是避免對每筆記錄執行 I/O。為達成這個目標，Dataflow 會使用 Bloom 篩選器和垃圾收集。

Bloom 濾鏡

Bloom 篩選器是精簡的資料結構，可快速檢查集合成員資格。在 Dataflow 中，每個工作站都會保留所見每個 ID 的 Bloom 篩選器。當新的記錄 ID 送達時，工作人員會在篩選器中查詢該 ID。如果篩選器傳回 false，表示這筆記錄不是重複記錄，且工作人員不會在穩定儲存空間中查詢 ID。

Dataflow 會保留一組依時間分類的滾動式 Bloom 篩選器。記錄抵達時，Dataflow 會根據系統時間戳記挑選適當的篩選器進行檢查。這個步驟可防止 Bloom 篩選器在篩選器進行垃圾收集時飽和，並限制啟動時需要掃描的資料量。

垃圾收集

為避免記錄 ID 填滿儲存空間，Dataflow 會使用垃圾收集功能移除舊記錄。Dataflow 會使用系統時間戳記計算垃圾收集浮水印。

這項浮水印是根據在特定階段等待的實際時間長度而定。因此，這項工具也會提供管道中哪些部分速度緩慢的資訊。系統延遲時間指標就是以這類中繼資料為基礎，並顯示在 Dataflow 監控介面中。

如果記錄的時間戳記早於浮水印，且該時間的 ID 已遭垃圾收集，系統就會忽略該記錄。由於觸發垃圾收集作業的低浮水印要等到記錄傳送作業確認後才會前進，因此這些延遲抵達的記錄是重複的。

非決定性來源

Dataflow 會使用 Apache Beam SDK 將資料讀取至管道。如果處理作業失敗，Dataflow 可能會重試從來源讀取資料。在這種情況下，Dataflow 必須確保來源產生的每個不重複記錄都只會記錄一次。如果是 Pub/Sub Lite 或 Kafka 等確定性來源，系統會根據記錄的偏移量讀取記錄，因此不需要執行這個步驟。

由於 Dataflow 無法自動指派記錄 ID，因此非決定性來源必須告知 Dataflow 記錄 ID，以免重複。如果來源為每筆記錄提供專屬 ID，連接器會在管道中使用隨機排序，移除重複項目。系統會篩除 ID 相同的記錄。 如要瞭解 Dataflow 如何在使用 Pub/Sub 做為來源時，實作僅須處理一次的作業，請參閱「使用 Pub/Sub 進行串流」頁面中的「僅須處理一次的作業」一節。

在管道中執行自訂 DoFn 時，Dataflow 無法保證每個記錄只會執行一次這段程式碼。為確保在工作站發生故障時至少處理一次記錄，Dataflow 可能會多次透過轉換執行特定記錄，或在多個工作站上同時執行同一筆記錄。如果在管道中加入程式碼，與外部服務聯絡等動作可能會針對特定記錄執行多次。

如要讓非決定性處理程序有效成為決定性程序，請使用檢查點。使用檢查點時，轉換的每個輸出內容都會先檢查點化至穩定儲存空間，並取得專屬 ID，然後才會傳送至下一個階段。Dataflow 的隨機傳送重試作業會轉送已檢查點的輸出內容。雖然程式碼可能會多次執行，但 Dataflow 會確保只儲存其中一次執行的輸出內容。Dataflow 使用一致的儲存空間，防止重複項目寫入穩定儲存空間。

僅傳送一次輸出內容

Apache Beam SDK 內建接收器，可確保不會產生重複項目。請盡可能使用其中一個內建接收器。

如要自行編寫接收器，最佳做法是讓函式物件成為冪等，這樣一來，函式物件就能視需要盡量重試，而不會造成非預期的副作用。不過，實作接收器功能的轉換通常會有不確定的元件，如果重試，可能會發生變化。

舉例來說，在視窗式匯總中，視窗中的記錄集可能不具決定性。具體來說，視窗可能會嘗試使用元素 e0、e1、e2 觸發事件。工作站可能會在提交視窗處理程序前當機，但不會在這些元素以副作用形式傳送前當機。工作人員重新啟動時，視窗會再次觸發，並收到延遲的元素 e3。由於這個元素會在視窗提交前抵達，因此不會計為延遲資料，所以系統會使用元素 e0、e1、e2、e3 再次呼叫 DoFn。這些元素隨後會傳送至副作用作業。在此情境中，冪等性沒有幫助，因為每次傳送的邏輯記錄集都不同。

如要解決 Dataflow 中的非決定性問題，請使用內建的 Reshuffle 轉換。Dataflow 隨機重組資料時，會將資料寫入持久性儲存空間，因此如果作業在隨機重組後重試，任何非決定性產生的元素都會保持穩定。使用 Reshuffle 轉換作業可確保只有一個版本的 DoFn 輸出內容能通過隨機播放邊界。下列模式可確保副作用作業一律會收到確定性記錄以輸出：

c.apply(Window.<..>into(FixedWindows.of(Duration.standardMinutes(1))))
 .apply(GroupByKey.<..>.create())
 .apply(new PrepareOutputData())
 .apply(Reshuffle.<..>of())
 .apply(WriteToSideEffect());

為確保 Dataflow 執行器知道元素必須穩定，才能執行 DoFn，請將 RequiresStableInput 註解新增至 DoFn。

瞭解詳情