升級串流管道

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

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

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

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

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

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

最佳做法

• 請分開升級 Apache Beam SDK 版本，不要一併變更管道程式碼。
• 每次變更後，請先測試管道，再進行其他更新。
• 定期升級管道使用的 Apache Beam SDK 版本。
• 盡可能使用自動化方法，例如在飛行期間更新，或自動平行管道更新。
• 盡可能使用受管理 I/O，享有自動升級連接器版本的優點。

在飛行途中更新

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

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

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

如需執行中工作更新的操作說明，請參閱「更新現有管道」。

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

使用範本 (傳統範本、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 選項。設定並行時間長度會觸發自動並行管道更新工作流程，

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

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

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

--dataflowServiceOptions="parallel_replace_job_cancel_on_drain_timeout=false"

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

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

--dataflow_service_options="parallel_replace_job_cancel_on_drain_timeout=false"

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

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

--dataflow_service_options="parallel_replace_job_cancel_on_drain_timeout=false"

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

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

--additional-experiments="parallel_replace_job_cancel_on_drain_timeout=false"

--additional-experiments="create_or_update_job"

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"
]

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 設定：
  • 指出是否要在舊工作耗盡資源時，預先為新工作佈建 worker。值：true (預設) 或 false。如果是 Terraform 和 Config Connector，建議將此選項設為 true，以免資源佈建逾時。如果 parallel_replace_job_preallocate_compute_resources 設為 false，新工作會維持待處理狀態，直到舊工作排空為止。

使用 Pub/Sub 快照和搜尋功能重新處理訊息

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

在開發和測試期間，您也可以使用 Pub/Sub Seek 重複播放已知訊息，驗證管道的輸出內容。使用 Pub/Sub Seek 時，如果管道正在取用訂閱項目，請勿搜尋訂閱項目快照。如果這麼做，搜尋作業可能會使 Dataflow 的水位線邏輯失效，並影響 Pub/Sub 訊息的僅一次處理作業。

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

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 是 管道 A 的浮水印。 為求簡化，假設浮水印完美無缺，沒有延遲資料。水平軸代表處理時間和 wall time。這兩個管道都使用五分鐘的固定 (滾動) 視窗。浮水印通過每個視窗的結尾後，系統就會觸發結果。

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

限制

使用自動或手動平行管道更新時，有下列限制：

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

自動平行更新管道

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

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

自動化平行管道服務選項

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

--additional-experiments="create_or_update_job"

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"
]

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 會等待所有 worker 完成佈建，再開始處理資料。如要監控部署狀態，請查看 Dataflow 工作記錄。

手動執行平行管道

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

處理重複的輸出內容

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

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

   

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

   

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

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

   

4. 此時只有「Pipeline B」正在執行。您可以從 BigQuery 檢視畫面 (Façade View) 查詢，該檢視畫面會做為資料表 A 和資料表 B 的外觀。如果兩個資料表中的資料列具有相同時間戳記，請將檢視畫面設為傳回「資料表 B」中的資料列，或在「資料表 B」中沒有資料列時，改為傳回「資料表 A」中的資料列。下圖顯示從「資料表 A」和「資料表 B」讀取資料的檢視畫面 (立面檢視畫面)。

   

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

如果新管道部署作業偵測到問題，平行管道可簡化回溯作業。在本例中，您可能想讓「Pipeline A」持續執行，同時監控「Pipeline 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 合併至主資料表。

後續步驟

• 如需詳細步驟，請參閱「更新現有管道」。