升級串流管道

本頁提供升級串流管道的指引和建議。舉例來說，您可能需要升級至新版 Apache Beam SDK，或更新管道程式碼。我們提供不同選項，以因應各種情況。

批次管道會在工作完成時停止，而串流管道通常會持續執行，以提供不間斷的處理作業。因此，升級串流管道時，請考量下列事項：

• 您可能需要盡量減少或避免管道中斷。在某些情況下，部署新版管道時，您或許可以容許暫時中斷處理作業。在其他情況下，應用程式可能無法容忍任何中斷。
• 管道更新程序必須處理結構定義變更，盡量減少對訊息處理和其他附加系統的干擾。舉例來說，如果事件處理管道中的訊息結構定義有所變更，下游資料接收器可能也需要變更結構定義。

視管道和更新需求而定，您可以使用下列其中一種方法更新串流管道：

• 執行飛行中更新
• 啟動替換工作
• 執行平行管道

如要進一步瞭解更新期間可能發生的問題，以及如何避免這些問題，請參閱「驗證替代工作」和「工作相容性檢查」。

最佳做法

• 請分開升級 Apache Beam SDK 版本，並變更管道程式碼。
• 每次變更後，請先測試管道，再進行其他更新。
• 定期升級管道使用的 Apache Beam SDK 版本。
• 盡可能使用自動化方法，例如更新進行中的作業或自動平行管道更新。
• 盡可能使用受管理 I/O，以享有自動升級連接器版本的好處。

執行飛行中更新

您可以更新部分進行中的串流管道，而不必停止作業。這種情況稱為「執行中工作更新」。只有在特定情況下，才能在飛行期間更新工作：

• 工作必須使用 Streaming Engine。
• 工作必須處於執行中狀態。
• 您只會變更工作使用的工作站數量。

詳情請參閱「設定自動調度資源範圍」一文。

如需瞭解如何執行執行中作業更新，請參閱「更新現有管道」。

自動建立或更新 (新增或更新) 範本

使用範本 (傳統範本、Flex 範本、Terraform 或 Config Connector) 啟動管道時，可以透過 create_or_update_job 實驗功能使用建立或更新 (upsert) 功能。

在 additional_experiments 參數或 additional-experiments 標記中指定 create_or_update_job 時：

• 如果指定工作名稱的執行中或排空工作已存在，範本服務會自動啟動新工作，做為現有工作的更新。
• 如果沒有名稱相同的進行中工作，範本服務會啟動新工作，視為建立新工作。

這項實驗可讓您在啟動範本時，不必以程式輔助方式判斷是否要使用建立或更新 API 動作。

如要查看使用這項實驗的 Terraform 和 Config Connector 程式碼範例，請參閱下列章節：

• 傳送自動停止和更換更新要求
• 傳送自動平行管道更新要求

啟動替換工作

如果更新後的工作與現有工作相容，您可以使用 update 選項更新管道。替換現有工作時，新工作會執行更新後的管道程式碼。Dataflow 服務會保留工作名稱，但會使用更新後的工作 ID 執行替換工作。這個程序可能會導致停機，因為現有工作會停止，相容性檢查會執行，新工作也會啟動。詳情請參閱「替換工作造成的影響」。

Dataflow 會執行相容性檢查，確保更新後的管道程式碼能安全部署至執行中的管道。某些程式碼變更會導致相容性檢查失敗，例如在現有步驟中新增或移除側邊輸入。如果相容性檢查失敗，您就無法執行就地工作更新。

如需啟動替代工作的操作說明，請參閱「啟動替代工作」。

如果管道更新與目前的工作不相容，您需要停止並替換管道。如果管道無法容忍停機時間，請執行平行管道。

手動停止並更換

如要手動停止及替換管道，請取消或排除管道，然後替換為更新後的管道。取消管道會導致 Dataflow 立即停止處理作業，並盡快關閉資源，這可能會導致部分處理中的資料 (又稱傳輸中資料) 遺失。為避免資料遺失，在大多數情況下，建議採取排除管道的動作。您也可以使用 Dataflow 快照儲存串流管道的狀態，這樣就能啟動新版 Dataflow 作業，且不會遺失狀態。詳情請參閱「使用 Dataflow 快照」。

管道排除作業會立即關閉任何處理中的時間區間，並啟動所有觸發條件。雖然傳輸中的資料不會遺失，但排除作業可能會導致時間區間的資料不完整。如果發生這種情況，處理中的時間區間會發出部分或不完整的結果。詳情請參閱「排除作業對工作造成的影響」。現有工作完成後，請啟動新的串流工作，其中包含更新後的管道程式碼，以便繼續處理作業。

使用這種方法時，現有串流工作停止和替換管道準備好繼續處理資料之間，會有一段停機時間。不過，取消或排空現有管道，然後使用更新後的管道啟動新工作，比執行平行管道簡單。

詳情請參閱「排空 Dataflow 工作」。排空目前的工作後，請啟動名稱相同的新工作。

自動停止並更換

Dataflow 提供 API，支援啟動自動停止並取代更新。這種宣告式工作流程可免除手動程序步驟。您宣告要取代的工作，新工作就會啟動並自動協調轉換作業。

使用這個工作流程時，系統會在舊工作仍在執行時，佈建新的工作資源。接著，舊工作會自動收到排空信號。舊工作排空完畢或達到使用者指定的逾時時間後，新工作會立即開始處理資料。如果管道無法容忍重複資料或部分匯總，但可以接受舊工作排空時的短暫處理暫停，請使用這個工作流程。

傳送自動停止和更換更新要求

如要使用這項工作流程，請按照下列指示操作：

• 您必須設定 parallel_replace_job_max_stop_duration 選項。
• 您不得設定 parallel_replace_job_min_parallel_pipelines_duration 選項。設定平行時間長度會觸發自動平行管道更新工作流程。

使用下列服務選項，啟動自動停止和更換更新要求：

Java

方法 1：使用相同的工作名稱更新

注意：如果使用相同的工作名稱更新 (選項 1)，新工作必須在先前工作開始至少兩分鐘後啟動。否則更新就會失敗並產生錯誤。這項限制可避免因用戶端程式庫重試或過時的遠端程序呼叫，而導致重複的平行更新。

--update \
--dataflowServiceOptions="update_strategy_parallel_job_update" \
--dataflowServiceOptions="parallel_replace_job_max_stop_duration=DURATION"

• 如要使用相同名稱執行自動停止及取代更新，請使用 --update 標記和 update_strategy_parallel_job_update 選項。
• 如要改為執行就地更新，請使用 update_strategy_in_place_update。

方法 2：使用其他職稱更新

--dataflowServiceOptions="parallel_replace_job_name=OLD_JOB_NAME" \
--dataflowServiceOptions="parallel_replace_job_max_stop_duration=DURATION"

• 如要依 ID 指定舊工作，而非依工作名稱，請使用 --dataflowServiceOptions="parallel_replace_job_id=OLD_JOB_ID"。
• 如果您指定新的工作名稱並使用 --update 旗標，Dataflow 會搜尋具有新名稱的現有工作，導致發生錯誤。

選用：停用自動取消功能

指定 parallel_replace_job_max_stop_duration 選項時，系統預設會啟用自動取消功能。如要停用自動取消功能，請將 parallel_replace_job_cancel_on_drain_timeout 選項設為 false。

--dataflowServiceOptions="parallel_replace_job_cancel_on_drain_timeout=false"

如果停用自動取消功能，且舊工作停滯在排空狀態，舊工作和新工作會繼續並行執行。

Python

方法 1：使用相同的工作名稱更新

注意：如果使用相同的工作名稱更新 (選項 1)，新工作必須在先前工作開始至少兩分鐘後啟動。否則更新就會失敗並產生錯誤。這項限制可避免因用戶端程式庫重試或過時的遠端程序呼叫，而導致重複的平行更新。

--update \
--dataflow_service_options="update_strategy_parallel_job_update" \
--dataflow_service_options="parallel_replace_job_max_stop_duration=DURATION"

• 如要使用相同名稱執行自動停止及取代更新，請使用 --update 標記和 update_strategy_parallel_job_update 選項。
• 如要改為執行就地更新，請使用 update_strategy_in_place_update。

方法 2：使用其他職稱更新

--dataflow_service_options="parallel_replace_job_name=OLD_JOB_NAME" \
--dataflow_service_options="parallel_replace_job_max_stop_duration=DURATION"

• 如要依 ID 指定舊工作，而非依工作名稱，請使用 --dataflow_service_options="parallel_replace_job_id=OLD_JOB_ID"。
• 如果您指定新的工作名稱並使用 --update 旗標，Dataflow 會搜尋具有新名稱的現有工作，導致發生錯誤。

選用：停用自動取消功能

指定 parallel_replace_job_max_stop_duration 選項時，系統預設會啟用自動取消功能。如要停用自動取消功能，請將 parallel_replace_job_cancel_on_drain_timeout 選項設為 false。

--dataflow_service_options="parallel_replace_job_cancel_on_drain_timeout=false"

如果停用自動取消功能，且舊工作停滯在排空狀態，舊工作和新工作會繼續並行執行。

Go

方法 1：使用相同的工作名稱更新

注意：如果使用相同的工作名稱更新 (選項 1)，新工作必須在先前工作開始至少兩分鐘後啟動。否則更新就會失敗並產生錯誤。這項限制可避免因用戶端程式庫重試或過時的遠端程序呼叫，而導致重複的平行更新。

--update \
--dataflow_service_options="update_strategy_parallel_job_update" \
--dataflow_service_options="parallel_replace_job_max_stop_duration=DURATION"

• 如要使用相同名稱執行自動停止及取代更新，請使用 --update 標記和 update_strategy_parallel_job_update 選項。
• 如要改為執行就地更新，請使用 update_strategy_in_place_update。

方法 2：使用其他職稱更新

--dataflow_service_options="parallel_replace_job_name=OLD_JOB_NAME" \
--dataflow_service_options="parallel_replace_job_max_stop_duration=DURATION"

• 如要依 ID 指定舊工作，而非依工作名稱，請使用 --dataflow_service_options="parallel_replace_job_id=OLD_JOB_ID"。
• 如果您指定新的工作名稱並使用 --update 旗標，Dataflow 會搜尋具有新名稱的現有工作，導致發生錯誤。

選用：停用自動取消功能

指定 parallel_replace_job_max_stop_duration 選項時，系統預設會啟用自動取消功能。如要停用自動取消功能，請將 parallel_replace_job_cancel_on_drain_timeout 選項設為 false。

--dataflow_service_options="parallel_replace_job_cancel_on_drain_timeout=false"

如果停用自動取消功能，且舊工作停滯在排空狀態，舊工作和新工作會繼續並行執行。

gcloud

方法 1：使用相同的工作名稱更新

注意：如果使用相同的工作名稱更新 (選項 1)，新工作必須在先前工作開始至少兩分鐘後啟動。否則更新就會失敗並產生錯誤。這項限制可避免因用戶端程式庫重試或過時的遠端程序呼叫，而導致重複的平行更新。

--update \
--additional-experiments="update_strategy_parallel_job_update" \
--additional-experiments="parallel_replace_job_max_stop_duration=DURATION"

• 如要使用相同名稱執行自動停止及取代更新，請使用 --update 標記和 update_strategy_parallel_job_update 選項。
• 如要改為執行就地更新，請使用 update_strategy_in_place_update。

方法 2：使用其他職稱更新

--additional-experiments="parallel_replace_job_name=OLD_JOB_NAME" \
--additional-experiments="parallel_replace_job_max_stop_duration=DURATION"

• 如要依 ID 指定舊工作，而非依工作名稱，請使用 --additional-experiments="parallel_replace_job_id=OLD_JOB_ID"。
• 如果您指定新的工作名稱並使用 --update 旗標，Dataflow 會搜尋具有新名稱的現有工作，導致發生錯誤。

選用：停用自動取消功能

指定 parallel_replace_job_max_stop_duration 選項時，系統預設會啟用自動取消功能。如要停用自動取消功能，請將 parallel_replace_job_cancel_on_drain_timeout 選項設為 false。

--additional-experiments="parallel_replace_job_cancel_on_drain_timeout=false"

如果停用自動取消功能，且舊工作停滯在排空狀態，舊工作和新工作會繼續並行執行。

選用：新增或更新 (建立或更新工作)

如要啟用 upsert (建立或更新工作) 行為，請按照下列步驟操作：

--additional-experiments="create_or_update_job"

Terraform

additional_experiments = [
  "parallel_replace_job_max_stop_duration=DURATION",
  "parallel_replace_job_cancel_on_drain_timeout=true",
  "update_strategy_parallel_job_update",
  "parallel_replace_job_preallocate_compute_resources=true",
  "create_or_update_job"
]

Config Connector

metadata:
  annotations:
    # Force to use only direct controller. Multiple controllers can cause a Dataflow job to enter into a continuous update loop.
    # https://docs.cloud.google.com/config-connector/docs/concepts/controller-types#underlying-controller-types
    alpha.cnrm.cloud.google.com/reconciler: direct
    # Optional but recommended: Dataflow batch jobs do not support the drain operation. But for Dataflow streaming jobs, prefer "drain" over "cancel" as an on-delete option.
    cnrm.cloud.google.com/on-delete: drain
    # Optional but recommended: Set deletion-policy to "abandon" to avoid accidental deletion, this will ignore the on-delete option.
    cnrm.cloud.google.com/deletion-policy: abandon
spec:
  ...
  additionalExperiments:
    - "parallel_replace_job_max_stop_duration=DURATION"
    - "parallel_replace_job_cancel_on_drain_timeout=true"
    - "update_strategy_parallel_job_update"
    - "parallel_replace_job_preallocate_compute_resources=true"
    - "create_or_update_job"

替換下列變數：

• 您必須提供 parallel_replace_job_name 或 parallel_replace_job_id，以識別要取代的工作：
  • OLD_JOB_NAME：要取代的工作名稱。
  • OLD_JOB_ID：要取代的工作 ID。
• 您必須提供 parallel_replace_job_max_stop_duration 值，才能啟用自動停止和更換功能：
  • DURATION：新工作等待舊工作排空的最多時間。時間長度必須格式化為以 s、m 或 h 結尾的字串 (例如 30m、1h)。
• 使用這個工作流程時，請勿設定 parallel_replace_job_min_parallel_pipelines_duration 選項。設定這個選項會觸發自動平行管道更新工作流程。
• 選用：設定 parallel_replace_job_cancel_on_drain_timeout 選項。由於系統預設會啟用自動取消功能 (預設為 true)，因此設定 parallel_replace_job_max_stop_duration 選項時，不需要明確設定這個選項來啟用自動取消功能。
  • 如要保留預設行為，請省略這個選項或將其設為 true。
  • 如要停用自動取消功能，請將這個選項設為 false。如果將這個選項設為 false，且舊工作卡在排空狀態，舊工作和新工作都會繼續並行執行。
• 選用：定義 parallel_replace_job_preallocate_compute_resources 設定：
  • 指定是否要在舊工作排空時，預先為新工作佈建工作站。值：true (預設) 或 false。對於 Terraform 和 Config Connector，建議將這個選項設為 true，以免資源佈建逾時。如果將 parallel_replace_job_preallocate_compute_resources 設為 false，新工作會維持待處理狀態，直到舊工作排空為止。

使用 Pub/Sub 快照和 Seek 重新處理訊息

在某些情況下，更換或取消耗盡的管道後，您可能需要重新處理先前傳送的 Pub/Sub 訊息。舉例來說，您可能需要使用更新的商業邏輯重新處理資料。Pub/Sub 搜尋功能可讓您重播 Pub/Sub 快照中的訊息。您可以搭配使用 Pub/Sub Seek 與 Dataflow，從建立訂閱項目快照的時間開始重新處理訊息。

在開發和測試期間，您也可以使用 Pub/Sub Seek 重複播放已知訊息，驗證管道的輸出內容。使用 Pub/Sub Seek 時，請勿在管道耗用訂閱項目時，搜尋訂閱項目快照。否則，搜尋作業可能會使 Dataflow 的浮水印邏輯失效，並影響 Pub/Sub 訊息的單次處理作業。

建議您在終端機視窗中，按照下列 gcloud CLI 工作流程，搭配使用 Pub/Sub Seek 與 Dataflow 管道：

1. 如要建立訂閱項目的快照，請使用 gcloud pubsub snapshots create 指令：

   gcloud pubsub snapshots create SNAPSHOT_NAME --subscription=PIPELINE_SUBSCRIPTION_NAME
   

2. 如要排空或取消管道，請使用 gcloud dataflow jobs drain 或 gcloud dataflow jobs cancel 指令：

   gcloud dataflow jobs drain JOB_ID
   

   或

   gcloud dataflow jobs cancel JOB_ID
   

3. 如要搜尋快照，請使用 gcloud pubsub subscriptions seek 指令：

   gcloud pubsub subscriptions seek SNAPSHOT_NAME
   

4. 部署會取用訂閱項目的新管道。

同時執行多個管道

如要避免更新期間串流管道中斷，可以平行執行管道。這種做法可讓您使用更新後的管道程式碼啟動新的串流工作，並與現有工作並行執行。您可以使用 Dataflow 的自動平行管道更新部署工作流程，也可以手動執行這些步驟。

平行管道總覽

建立新管道時，請使用與現有管道相同的開窗策略。如果是手動工作流程，請讓現有管道繼續執行，直到其浮水印超過更新後管道處理的最早完整視窗時間戳記為止，然後排除或取消現有管道。如果是自動化工作流程，系統會為您完成這項工作。更新後的管道會繼續執行，並有效接管處理作業。

下圖說明了這個程序。

在圖中，管道 B 是接手管道 A 的更新工作。t 值是 Pipeline B 處理的最早完整視窗的時間戳記。值 w 是 Pipeline A 的浮水印。 為求簡單，我們假設浮水印完美無缺，沒有遲到的資料。水平軸代表處理時間和 wall time。這兩個管道都使用五分鐘的固定 (翻滾) 時間區間。浮水印超過每個時間區間的結尾時，系統就會觸發結果。

由於並行輸出發生在兩個管道重疊的時間範圍內，請將兩個管道設定為將結果寫入不同的目的地。下游系統隨後可以使用兩個目的地接收器的抽象化 (例如資料庫檢視畫面)，查詢合併結果。這些系統也可以使用抽象化，從重疊期間中移除重複的結果。詳情請參閱「處理重複的輸出內容」。

限制

使用自動或手動並行管道更新時，會受到下列限制：

• 僅限自動更新：新的平行工作必須是 Streaming Engine 工作。
• 系統不允許名稱相同的並行作業。不過，使用相同作業名稱執行自動停止及取代或平行管道更新時，可以重複使用作業名稱。在這種情況下，新作業必須在前一個作業開始至少兩分鐘後啟動。這項限制可防止因重複的用戶端程式庫重試或過時的遠端程序呼叫，導致多個平行更新。
• 在相同輸入內容上平行執行兩個管道，可能會導致資料重複、部分彙整，以及將資料插入接收器時可能發生排序問題。下游系統必須經過設計，才能預期及管理這些結果。
• 從 Pub/Sub 來源讀取資料時，不建議多個管道使用相同的訂閱項目，否則可能會導致正確性問題。不過，在某些用途中 (例如擷取、轉換、載入 (ETL) 管道)，在兩個管道中使用相同訂閱項目可能會減少重複作業。只要您為重疊時間長度提供非零值，就可能發生自動調度資源問題。使用執行中的工作更新功能即可減輕影響。詳情請參閱「微調 Pub/Sub 串流管道的自動調度資源功能」。
• 如果是 Apache Kafka，您可以啟用 Kafka 中的偏移量提交，盡量減少重複資料。如要在 Kafka 中啟用偏移量提交，請參閱「提交回 Kafka」一文。

自動更新平行管道

Dataflow 提供 API，可啟動平行取代工作。這個宣告式 API 會將執行程序步驟的手動工作抽象化。您只要宣告要更新的工作，新工作就會與舊工作平行執行。新工作執行指定時間後，舊工作就會排空。這項功能可避免更新期間處理作業暫停，也能減少更新不相容管道所需的操作工作。

如果管道可容許部分重複或部分匯總，且插入資料時不需要嚴格排序，這個更新方法就非常適合。這類方法適用於 ETL 管道，以及使用「至少一次串流模式」和 Redistribute 轉換，且允許重複項目設為 true 的管道。

自動化平行管道服務選項

注意： 無論您選擇哪種更新策略 (選項 1：使用相同的工作名稱更新，或選項 2：使用不同的工作名稱更新)，都必須設定 parallel_replace_job_min_parallel_pipelines_duration 選項。

如要自動平行更新管道，請使用下列服務選項：

服務選項	選用或必要	說明	依附元件或排除項目
update_strategy_parallel_job_update	必要 (方法 1：使用相同的工作名稱更新)	執行平行更新的指令，在以相同工作名稱更新時，會同時執行兩個管道，盡量縮短停機時間。	必須與 --update 旗標和 parallel_replace_job_min_parallel_pipelines_duration 一併設定。
update_strategy_in_place_update	選用	平行更新的替代方案。執行標準的就地工作更新。	必須與 --update 旗標一併設定。 與 update_strategy_parallel_job_update 互斥。 設定這個選項後，系統會忽略其他與平行工作相關的選項。
parallel_replace_job_min_parallel_pipelines_duration	必要	指定兩個管道同時執行的最短時間。經過這段時間後，系統會將排空信號傳送至舊工作。可接受的值範圍為 0s (建議用於零重疊) 至 744h (31 天)。	必須搭配舊職位目標對象設定。下列其中一項： 方法 1 - 使用相同的工作名稱： update_strategy_parallel_job_update，或 方法 2 - 使用其他工作名稱： parallel_replace_job_name (或使用 parallel_replace_job_id 識別工作)
parallel_replace_job_name 或 parallel_replace_job_id (擇一)	必要 (方法 2：使用其他工作名稱更新)	透過名稱或 ID 識別要替換的舊工作，以便在更新名稱時使用。	必須設定「parallel_replace_job_min_parallel_pipelines_duration」。 請勿搭配使用 --update 旗標或 parallel_replace_job_id。
parallel_replace_job_max_stop_duration	選用	舊工作在觸發自動取消前，可排空的最長時間。例如 30m 或 1h。	必須設定平行更新工作流程 (選項 1 或選項 2)。
parallel_replace_job_cancel_on_drain_timeout	選填 如果設定了停止時間上限，預設值為 true。	布林值選項，用於指定舊工作排空時間超過 parallel_replace_job_max_stop_duration 時是否要取消。	須搭配 parallel_replace_job_max_stop_duration 使用。 設為 false 即可停用自動取消功能。如果停用自動取消功能，且舊工作停滯在排空狀態，舊工作和新工作就會繼續並行執行。

傳送自動平行管道更新要求

如要使用自動化工作流程，請啟動新的串流工作。您可以更新工作，並使用相同或不同的工作名稱。

Java

方法 1：使用相同的工作名稱更新

注意：如果使用相同的工作名稱更新 (選項 1)，新工作必須在先前工作開始至少兩分鐘後啟動。否則更新就會失敗並產生錯誤。這項限制可避免因用戶端程式庫重試或過時的遠端程序呼叫，而導致重複的平行更新。

--update \
--dataflowServiceOptions="update_strategy_parallel_job_update" \
--dataflowServiceOptions="parallel_replace_job_min_parallel_pipelines_duration=DURATION"

• 如要使用相同名稱執行平行更新，請使用 --update 旗標和 update_strategy_parallel_job_update 選項。
• 如要執行就地更新，且不移除與平行工作相關的選項，請使用 update_strategy_in_place_update，而非 update_strategy_parallel_job_update。

方法 2：使用其他職稱更新

--dataflowServiceOptions="parallel_replace_job_name=OLD_JOB_NAME" \
--dataflowServiceOptions="parallel_replace_job_min_parallel_pipelines_duration=DURATION"

• 如要依 ID 指定舊工作，而非依工作名稱，請使用 --dataflowServiceOptions="parallel_replace_job_id=OLD_JOB_ID"。
• 如果您指定新的工作名稱並使用 --update 旗標，Dataflow 會搜尋具有新名稱的現有工作，導致發生錯誤。

選用：設定排除逾時和自動取消

您可以在任一設定中附加下列選項，設定排空逾時時間，並在舊工作停滯時自動取消。

--dataflowServiceOptions="parallel_replace_job_max_stop_duration=DRAIN_TIMEOUT_DURATION" \
--dataflowServiceOptions="parallel_replace_job_cancel_on_drain_timeout=true"

如果停用自動取消功能，且舊工作停滯在排空狀態，舊工作和新工作會繼續並行執行。

Python

方法 1：使用相同的工作名稱更新

注意：如果使用相同的工作名稱更新 (選項 1)，新工作必須在先前工作開始至少兩分鐘後啟動。否則更新就會失敗並產生錯誤。這項限制可避免因用戶端程式庫重試或過時的遠端程序呼叫，而導致重複的平行更新。

--update \
--dataflow_service_options="update_strategy_parallel_job_update" \
--dataflow_service_options="parallel_replace_job_min_parallel_pipelines_duration=DURATION"

• 如要使用相同名稱執行平行更新，請使用 --update 旗標和 update_strategy_parallel_job_update 選項。
• 如要執行就地更新，且不移除與平行工作相關的選項，請使用 update_strategy_in_place_update，而非 update_strategy_parallel_job_update。

方法 2：使用其他職稱更新

--dataflow_service_options="parallel_replace_job_name=OLD_JOB_NAME" \
--dataflow_service_options="parallel_replace_job_min_parallel_pipelines_duration=DURATION"

• 如要依 ID 指定舊工作，而非依工作名稱，請使用 --dataflow_service_options="parallel_replace_job_id=OLD_JOB_ID"。
• 如果您指定新的工作名稱並使用 --update 旗標，Dataflow 會搜尋具有新名稱的現有工作，導致發生錯誤。

選用：設定排除逾時和自動取消

您可以在任一設定中附加下列選項，設定排空逾時時間，並在舊工作停滯時自動取消。

--dataflow_service_options="parallel_replace_job_max_stop_duration=DRAIN_TIMEOUT_DURATION" \
--dataflow_service_options="parallel_replace_job_cancel_on_drain_timeout=true"

如果停用自動取消功能，且舊工作停滯在排空狀態，舊工作和新工作會繼續並行執行。

Go

方法 1：使用相同的工作名稱更新

注意：如果使用相同的工作名稱更新 (選項 1)，新工作必須在先前工作開始至少兩分鐘後啟動。否則更新就會失敗並產生錯誤。這項限制可避免因用戶端程式庫重試或過時的遠端程序呼叫，而導致重複的平行更新。

--update \
--dataflow_service_options="update_strategy_parallel_job_update" \
--dataflow_service_options="parallel_replace_job_min_parallel_pipelines_duration=DURATION"

• 如要使用相同名稱執行平行更新，請使用 --update 旗標和 update_strategy_parallel_job_update 選項。
• 如要執行就地更新，且不移除與平行工作相關的選項，請使用 update_strategy_in_place_update，而非 update_strategy_parallel_job_update。

方法 2：使用其他職稱更新

--dataflow_service_options="parallel_replace_job_name=OLD_JOB_NAME" \
--dataflow_service_options="parallel_replace_job_min_parallel_pipelines_duration=DURATION"

• 如要依 ID 指定舊工作，而非依工作名稱，請使用 --dataflow_service_options="parallel_replace_job_id=OLD_JOB_ID"。
• 如果您指定新的工作名稱並使用 --update 旗標，Dataflow 會搜尋具有新名稱的現有工作，導致發生錯誤。

選用：設定排除逾時和自動取消

您可以在任一設定中附加下列選項，設定排空逾時時間，並在舊工作停滯時自動取消。

--dataflow_service_options="parallel_replace_job_max_stop_duration=DRAIN_TIMEOUT_DURATION" \
--dataflow_service_options="parallel_replace_job_cancel_on_drain_timeout=true"

如果停用自動取消功能，且舊工作停滯在排空狀態，舊工作和新工作會繼續並行執行。

gcloud

方法 1：使用相同的工作名稱更新

注意：如果使用相同的工作名稱更新 (選項 1)，新工作必須在先前工作開始至少兩分鐘後啟動。否則更新就會失敗並產生錯誤。這項限制可避免因用戶端程式庫重試或過時的遠端程序呼叫，而導致重複的平行更新。

--update \
--additional-experiments="update_strategy_parallel_job_update" \
--additional-experiments="parallel_replace_job_min_parallel_pipelines_duration=DURATION"

• 如要使用相同名稱執行平行更新，請使用 --update 旗標和 update_strategy_parallel_job_update 選項。
• 如要執行就地更新，且不移除與平行工作相關的選項，請使用 update_strategy_in_place_update，而非 update_strategy_parallel_job_update。

方法 2：使用其他職稱更新

--additional-experiments="parallel_replace_job_name=OLD_JOB_NAME" \
--additional-experiments="parallel_replace_job_min_parallel_pipelines_duration=DURATION"

• 如要依 ID 指定舊工作，而非依工作名稱，請使用 --additional-experiments="parallel_replace_job_id=OLD_JOB_ID"。
• 如果您指定新的工作名稱並使用 --update 標記，Dataflow 會搜尋具有新名稱的現有工作，導致發生錯誤。

選用：設定排除逾時和自動取消

您可以在任一設定中附加下列選項，設定排空逾時時間，並在舊工作停滯時自動取消。

--additional-experiments="parallel_replace_job_max_stop_duration=DRAIN_TIMEOUT_DURATION" \
--additional-experiments="parallel_replace_job_cancel_on_drain_timeout=true"

如果停用自動取消功能，且舊工作停滯在排空狀態，舊工作和新工作會繼續並行執行。

選用：新增或更新 (建立或更新工作)

如要啟用 upsert (建立或更新工作) 行為，請按照下列步驟操作：

--additional-experiments="create_or_update_job"

Terraform

additional_experiments = [
  "parallel_replace_job_min_parallel_pipelines_duration=DURATION",
  "parallel_replace_job_max_stop_duration=DRAIN_TIMEOUT_DURATION",
  "update_strategy_parallel_job_update",
  "create_or_update_job"
]

Config Connector

metadata:
  annotations:
    # Force to use only direct controller. Multiple controllers can cause a Dataflow job to enter into a continuous update loop.
    # https://docs.cloud.google.com/config-connector/docs/concepts/controller-types#underlying-controller-types
    alpha.cnrm.cloud.google.com/reconciler: direct
    # Optional but recommended: Dataflow batch jobs do not support the drain operation. But for Dataflow streaming jobs, prefer "drain" over "cancel" as an on-delete option.
    cnrm.cloud.google.com/on-delete: drain
    # Optional but recommended: Set deletion-policy to "abandon" to avoid accidental deletion, this will ignore the on-delete option.
    cnrm.cloud.google.com/deletion-policy: abandon
spec:
  ...
  additionalExperiments:
    - "parallel_replace_job_min_parallel_pipelines_duration=DURATION"
    - "parallel_replace_job_max_stop_duration=DRAIN_TIMEOUT_DURATION"
    - "update_strategy_parallel_job_update"
    - "create_or_update_job"

替換下列變數：

• 如果您使用其他工作名稱更新 (選項 2)，則必須提供 parallel_replace_job_name 或 parallel_replace_job_id，以識別要替換的工作。Terraform 或 Config Connector 不支援使用其他工作名稱更新。
  • OLD_JOB_NAME：要取代的工作名稱。
  • OLD_JOB_ID：要取代的工作 ID。

• DURATION：兩個管道平行執行的最短時間，以整數或浮點數表示。建議將時間設為 0s，確保不會重疊。時間一到，系統就會向舊工作傳送排空信號。

  時間長度必須介於 0 秒 (0s) 和 31 天 (744h) 之間。請使用 s、m 和 h 指定秒、分和小時。舉例來說，10m 是 10 分鐘。

• DRAIN_TIMEOUT_DURATION：選用。舊工作排空前，系統會等待多長時間，超過上限即會觸發自動取消。時間長度必須格式化為以 s、m 或 h 結尾的字串 (例如 30m、1h)。

• parallel_replace_job_cancel_on_drain_timeout：選用。是否要取消先前的工作 (如果工作在停止時間上限前未完成排空)。如果提供排空逾時時間長度，則預設值為 true。如要停用自動取消功能，請將此選項設為 false。如果將這個選項設為 false，且舊工作停滯在排空狀態，舊工作和新工作都會繼續並行執行。

啟動新工作時，Dataflow 會等待所有工作站完成佈建，再開始處理資料。如要監控部署狀態，請查看 Dataflow 工作記錄。

手動執行平行管道

如要處理更複雜的情況，或進一步控管更新程序，可以手動執行平行管道。讓現有管道繼續執行，直到浮水印超過更新後管道處理的最早完整視窗時間戳記為止。接著，排空或取消現有管道。

處理重複的輸出內容

以下範例說明處理重複輸出內容的方法。這兩個管道會將輸出內容寫入不同目的地，使用下游系統查詢結果，並從重疊期間中移除重複結果。這個範例使用的管道會從 Pub/Sub 讀取輸入資料、執行一些處理作業，然後將結果寫入 BigQuery。

1. 在初始狀態下，現有的串流管道 (管道 A) 正在執行，並使用訂閱項目 (訂閱項目 A) 從 Pub/Sub 主題 (主題) 讀取訊息。結果會寫入 BigQuery 資料表 (資料表 A)。結果會透過 BigQuery 檢視畫面取用，該檢視畫面會做為門面，遮蓋基礎資料表變更。這個程序是外觀模式設計方法的應用。下圖顯示初始狀態。

   

2. 為更新後的管道建立新訂閱項目 (訂閱項目 B)。部署更新後的管道 (管道 B)，該管道會使用 訂閱項目 B 從 Pub/Sub 主題 (主題) 讀取資料，並寫入另一個 BigQuery 資料表 (資料表 B)。下圖說明瞭這個流程。

   

   此時，管道 A 和 管道 B 會平行執行，並將結果寫入不同的資料表。您會將時間 t 記錄為 管道 B 處理的最早完整視窗的時間戳記。

3. 當 管道 A 的浮水印超過時間 t 時，請排除 管道 A。排除管道時，所有開啟的視窗都會關閉，並完成處理中的資料。如果管道包含視窗，且完整視窗很重要 (假設沒有延遲資料)，請先讓兩個管道執行，直到您有完整的重疊視窗為止，再排除 管道 A。所有處理中的資料都處理完畢並寫入 資料表 A 後，請停止 管道 A 的串流工作。下圖顯示這個階段。

   

4. 此時，只有「管道 B」正在執行。您可以從 BigQuery 檢視區塊 (「門面檢視區塊」) 查詢，該檢視區塊會做為「資料表 A」和「資料表 B」的門面。如果兩個資料表中的資料列具有相同時間戳記，請將檢視區塊設為傳回「資料表 B」中的資料列；如果「資料表 B」中沒有資料列，則回溯至「資料表 A」。下圖顯示檢視區塊 (「門面檢視區塊」)從「資料表 A」和「資料表 B」讀取資料。

   

   此時，您可以刪除「訂閱方案 A」。

如果新管道部署作業偵測到問題，平行管道可簡化回溯作業。在這個範例中，您可能想讓管道 A 繼續執行，同時監控管道 B 是否正常運作。如果管道 B 發生任何問題，您可以復原為管道 A。

處理結構定義異動

資料處理系統通常需要隨著時間調整結構定義突變，有時是因應業務需求變化，有時是基於技術原因。套用結構定義更新時，通常需要審慎規劃及執行，以免業務資訊系統中斷運作。

假設管道會從 Pub/Sub 主題讀取含有 JSON 酬載的訊息。管道會將每則訊息轉換為 TableRow 執行個體，然後將資料列寫入 BigQuery 資料表。輸出資料表的結構定義與管道處理的訊息類似。在下圖中，結構定義稱為「結構定義 A」。

隨著時間推移，訊息結構定義可能會發生重大變化。舉例來說，新增、移除或取代欄位。結構定義 A 演變成新的結構定義。在接下來的討論中，新結構定義稱為「結構定義 B」。在這種情況下，管道 A 需要更新，且輸出資料表結構定義必須支援結構定義 B。

對於輸出資料表，您可以在不停機的情況下執行一些結構定義突變。舉例來說，您可以新增欄位或放寬資料欄模式，例如將 REQUIRED 變更為 NULLABLE，而不會發生停機情形。這些變動通常不會影響現有查詢。不過，如果結構定義突變，修改或移除現有結構定義欄位，就會導致查詢中斷或其他問題。以下方法可因應變更，且無須停機。

將管道寫入的資料分成主要資料表和一或多個暫存資料表。主要資料表會儲存管道寫入的歷來資料，暫存資料表則會儲存最新的管道輸出內容。您可以針對主要和暫存資料表定義 BigQuery 外觀檢視區塊，讓消費者查詢歷來和最新資料。

下圖修訂了先前的管道流程，納入暫存資料表 (暫存資料表 A)、主資料表和外觀檢視畫面。

在修訂後的流程中，管道 A 會處理使用結構定義 A 的訊息，並將輸出內容寫入具有相容結構定義的暫存資料表 A。主要資料表包含管道先前版本寫入的歷史資料，以及定期從暫存資料表合併的結果。消費者可以使用立面檢視畫面查詢最新資料，包括歷來資料和即時資料。

當訊息結構定義從「結構定義 A」變更為「結構定義 B」時，您可能需要更新管道程式碼，才能與使用「結構定義 B」的訊息相容。現有管道必須更新為新的實作方式。並行執行管道可確保串流資料處理作業持續運作，不會中斷。終止並更換管道會導致處理作業中斷，因為有一段時間沒有管道執行作業。

更新後的管道會寫入使用結構定義 B 的額外暫存資料表 (暫存資料表 B)。您可以使用協調式工作流程建立新的暫存資料表，然後更新管道。更新外觀檢視畫面，納入新暫存資料表的結果，可能需要使用相關工作流程步驟。

下圖顯示更新後的流程，其中包含具有「結構定義 B」的「暫存資料表 B」，以及如何更新外觀檢視畫面，以納入主要資料表和兩個暫存資料表的內容。

您可以定期或視需要將暫存資料表併入主要資料表，這項程序與管道更新程序不同。下圖顯示如何將暫存資料表 A 併入主要資料表。

後續步驟

• 如需更新現有管道的詳細步驟，請參閱這篇文章。