從 Pub/Sub 讀取資料到 Dataflow

本頁面說明在 Dataflow 中從 Pub/Sub 讀取資料的最佳做法。

Apache Beam 提供 Pub/Sub I/O 連接器的參考實作，供非 Dataflow 執行工具使用。不過，Dataflow 執行器會使用自訂的連接器實作。這項實作會利用 Google Cloud內部 API 和服務，提供低延遲浮水印、高浮水印準確度，以及高效率的簡化功能，確保訊息只處理一次。連接器適用於 Java、Python 和 Go。

僅需處理一次

Pub/Sub 會將事件發布者與事件消費者分離。應用程式會將訊息發布至主題，Pub/Sub 則會以非同步方式將訊息傳送給訂閱者。

Pub/Sub 會為成功發布至主題的每則訊息指派專屬訊息 ID。根據預設，Pub/Sub 會至少傳送一次訊息。為達到「至少一次」語意，如果 Pub/Sub 在確認期限內未收到訂閱者的確認，就會重試訊息傳遞。重試也可能在確認期限之前或訊息確認之後發生。

Dataflow 會在第一個融合階段成功處理訊息，並將處理作業的副作用寫入永久儲存空間後，確認訊息。為減少重複訊息數量，Dataflow 會在這個階段處理一批訊息時，持續延長確認期限。

由於 Pub/Sub 可能會重新傳送訊息，因此重複的訊息可能會送達管道。如果 Dataflow 管道使用一次性串流模式，Dataflow 會簡化這些訊息，以達到一次性語意。

如果管道可以容許部分重複記錄，請考慮改用至少一次串流模式。這個模式可大幅降低延遲時間和管道總成本。缺點是重複的訊息可能會處理兩次。詳情請參閱「選擇要使用的串流模式」。

依訊息屬性簡化

根據預設，Dataflow 會依據訊息 ID 簡化資料。不過，應用程式可能會以兩則不同的 Pub/Sub 訊息傳送同一筆記錄兩次。舉例來說，原始來源資料可能包含重複的記錄，或者應用程式可能誤將同一訊息發布兩次。如果確認訊息因網路問題或其他中斷情形而遭到捨棄，則重試可能會導致後者發生。在這種情況下，重複的訊息會有不同的訊息 ID。

視情況而定，資料可能包含可用於簡化的不重複欄位。例如，記錄可能包含不重複的交易 ID。您可以設定 Pub/Sub I/O 連接器，根據訊息屬性的值簡化訊息，而非使用 Pub/Sub 訊息 ID。只要發布者在重試期間一致設定這個屬性，Dataflow 就能偵測重複項目。訊息必須在 10 分鐘內發布至 Pub/Sub，才能簡化。

如要進一步瞭解如何使用 ID 屬性，請參閱下列 SDK 參考主題：

訂閱項目

設定管道時，您需要指定要從哪個 Pub/Sub 主題或 Pub/Sub 訂閱項目讀取資料。如果指定訂閱項目，請勿在多個管道使用相同的 Pub/Sub 訂閱項目。如果兩個管道從同一個訂閱項目讀取資料，每個管道都會以不確定的方式接收部分資料，這可能會導致訊息重複、浮水印延遲，以及自動調度資源效率不彰。請改為每個管道建立個別訂閱項目。

如果指定主題，連接器會建立新的臨時訂閱項目。每個管道都有專屬的訂閱項目。

時間戳記和浮水印

所有 Pub/Sub 訊息都有時間戳記，代表 Pub/Sub 收到訊息的時間。資料也可能包含「事件」時間戳記，也就是來源產生記錄的時間。

您可以設定連接器，從 Pub/Sub 訊息的屬性讀取事件時間戳記。在這種情況下，連接器會使用事件時間戳記做為浮水印。否則，系統預設會使用 Pub/Sub 訊息時間戳記。

如要進一步瞭解如何使用事件時間戳記，請參閱下列 SDK 參考主題：

Pub/Sub 連接器可存取 Pub/Sub 的私人 API，該 API 會提供訂閱項目中最舊未確認訊息的存在時間。這個 API 的延遲時間比 Cloud Monitoring 短。這項功能可讓 Dataflow 推進管道浮水印，並以低延遲時間輸出視窗化運算結果。

如果您將連接器設為使用事件時間戳記，Dataflow 會建立第二個 Pub/Sub 訂閱項目，稱為「追蹤訂閱項目」。Dataflow 會使用追蹤訂閱項目，檢查仍待處理的訊息事件時間。這種方法可讓 Dataflow 準確估算事件時間待處理項目。在包含追蹤訂閱項目的專案中，worker 服務帳戶至少須具備下列權限：

  • pubsub.subscriptions.create
  • pubsub.subscriptions.consume
  • pubsub.subscriptions.delete

此外，也需要 Pub/Sub 主題的 pubsub.topics.attachSubscription 權限。建議您建立自訂 Identity and Access Management 角色，並僅授予這些權限。

如要進一步瞭解浮水印，請參閱 StackOverflow 頁面，其中涵蓋 Dataflow 如何計算 Pub/Sub 浮水印。

如果管道有多個 Pub/Sub 來源，且其中一個來源的資料量非常低或處於閒置狀態，就會延遲整個浮水印的推進，增加整體管道延遲時間。如果管道中有根據浮水印設定的計時器或視窗匯總，這些也會延遲。

Pub/Sub 搜尋

Pub/Sub Seek 可讓使用者重播先前已確認的訊息。您可以搭配使用 Pub/Sub Seek 和 Dataflow，重新處理管道中的訊息。

不過，不建議在執行中的管道使用 Pub/Sub Seek。在執行中的管道中向後跳轉，可能會導致訊息重複或遭到捨棄。此外，這也會導致 Dataflow 的浮水印邏輯失效，並與包含已處理資料的管道狀態發生衝突。

如要使用 Pub/Sub Seek 重新處理訊息，建議採用下列工作流程：

  1. 建立訂閱項目的快照。
  2. 為 Pub/Sub 主題建立新的訂閱項目。新的訂閱項目會沿用快照。
  3. 排除或取消目前的 Dataflow 工作。
  4. 使用新的訂閱項目重新提交管道。

詳情請參閱「使用 Pub/Sub 快照和搜尋功能重新處理訊息」。

Pub/Sub 來源平行處理

Pub/Sub 來源會為每則訊息指派確定性金鑰，以便處理，並使用這些金鑰重組訊息。

  • 如果是 Streaming Engine 工作，Dataflow 會根據特定時間點的作用中 worker 數量，動態計算索引鍵數量：(number of workers * threads per worker) / 2。計算出的值會限制在 128 到 65,536 個金鑰之間，然後無條件捨去至最接近的 2 的次方 (小於或等於計算值的最大 2 的次方)。

  • 對於非 Streaming Engine 工作，鍵數為大於或等於 (4 * maximum workers) 的最小 2 的冪。

如要覆寫預設的重組金鑰數量，請設定 num_pubsub_keys 服務選項：

Java 

--dataflowServiceOptions=num_pubsub_keys=NUMBER_OF_KEYS

Python

--dataflow_service_options=num_pubsub_keys=NUMBER_OF_KEYS

Go

--dataflow_service_options=num_pubsub_keys=NUMBER_OF_KEYS

將 NUMBER_OF_KEYS 替換為金鑰數量。系統會使用大於或等於指定值的下一個 2 的冪。

舉例來說，您可以在下列情況設定這個選項：

如果設定這個選項，請考量「平行處理和分配」一節所述的取捨。

更新管道時，無法變更金鑰數量。如要變更現有管道工作的金鑰數量，必須啟動新工作。

不支援的 Pub/Sub 功能

Dataflow 執行器的 Pub/Sub I/O 連接器實作不支援下列 Pub/Sub 功能。

指數輪詢

建立 Pub/Sub 訂閱項目時，您可以設定使用指數輪詢重試政策。不過，指數輪詢不適用於 Dataflow。請改為使用「立即重試」重試政策建立訂閱項目。

否定確認或確認期限到期時，會觸發指數輪詢。不過，Dataflow 不會在管道程式碼失敗時傳送否定確認。而是無限期重試訊息處理作業，同時持續延長訊息的確認期限。

dead-letter 主題

請勿將 Pub/Sub dead-letter 主題與 Dataflow 搭配使用，原因如下：

  • Dataflow 會基於各種內部原因 (例如 worker 關閉) 傳送否定確認。因此，即使管道程式碼未發生任何失敗，訊息仍可能傳送至 dead-letter 主題。

  • 當第一項融合階段成功處理一批訊息後，Dataflow 會確認訊息。如果管道有多個融合階段，且在第一階段後的任何時間點發生失敗，訊息會視為已確認，不會傳送至 dead-letter 主題。

請改為在管道中明確實作無效信件模式，將失敗的訊息轉送至目的地，以便日後處理。部分 I/O 接收器內建支援無效信件佇列。下列範例會實作無效信件模式：

Pub/Sub「僅傳送一次」

Dataflow 有自己的「僅限一次」處理機制，因此不建議搭配使用 Pub/Sub 的「僅限一次」傳遞功能。啟用 Pub/Sub 僅限一次傳送功能會降低管道效能，因為這會限制可平行處理的訊息數量。

Pub/Sub 訊息排序

訊息排序是 Pub/Sub 的一項功能，可讓訂閱者按照發布順序接收訊息。

不建議搭配 Dataflow 使用訊息排序功能，原因如下：

  • Pub/Sub I/O 連接器可能不會保留訊息順序。
  • Apache Beam 並未定義元素處理順序的嚴格規範。因此，下游轉換可能不會保留順序。
  • 搭配使用 Pub/Sub 訊息排序和 Dataflow，可能會增加延遲時間並降低效能。

Pub/Sub 單一訊息轉換

單一訊息轉換 (SMT) 可讓您根據訊息在系統中串流時的屬性或資料，操控、驗證及篩選訊息。提供資料給 Dataflow 的訂閱項目不應使用會篩除訊息的 SMT，因為這可能會干擾自動調度資源。這是因為訂閱項目 SMT 篩選功能可能會導致待處理項目看起來比傳送至 Dataflow 的項目多，直到篩除的訊息實際由 SMT 處理為止。篩選訊息的主題 SMT 不會導致自動調度資源發生問題。

後續步驟