使用 Debezium 和 Pub/Sub (串流) 範本，透過變更資料擷取將資料從 MySQL 同步處理至 BigQuery

「使用 Debezium 和 Pub/Sub 從 MySQL 擷取變更資料並寫入 BigQuery」範本為串流管道，可從 MySQL 資料庫讀取含有變更資料的 Pub/Sub 訊息，然後將記錄寫入 BigQuery。Debezium 連接器會擷取 MySQL 資料庫的變更，並將變更後的資料發布至 Pub/Sub。範本接著會讀取 Pub/Sub 訊息，並將其寫入 BigQuery。

這個範本可用於同步處理 MySQL 資料庫和 BigQuery 資料表。管道會將變更的資料寫入 BigQuery 暫存資料表，並間歇性更新 BigQuery 資料表，複製 MySQL 資料庫。

管道相關規定

  • 必須部署 Debezium 連接器。
  • Pub/Sub 訊息必須以 Beam Row 序列化。

範本參數

必要參數

  • inputSubscriptions：要讀取的 Pub/Sub 輸入訂閱項目清單 (以逗號分隔)，格式為 <SUBSCRIPTION_NAME>,<SUBSCRIPTION_NAME>, ...。
  • changeLogDataset：用於儲存暫存資料表的 BigQuery 資料集，格式為 <DATASET_NAME>。
  • replicaDataset：BigQuery 資料集的位置，用於儲存副本資料表，格式為 <DATASET_NAME>。

選用參數

  • inputTopics：以逗號分隔的清單，列出 CDC：變更資料擷取資料會推送至的 Pub/Sub 主題。
  • updateFrequencySecs：管道更新 BigQuery 資料表 (複製 MySQL 資料庫) 的間隔。
  • useSingleTopic：如果將 Debezium 連接器設定為將所有資料表更新發布至單一主題，請將此值設為 true。預設值為：false。
  • useStorageWriteApi：如果為 true，管道會使用 BigQuery Storage Write API (https://cloud.google.com/bigquery/docs/write-api)。預設值為 false。詳情請參閱「使用 Storage Write API」(https://beam.apache.org/documentation/io/built-in/google-bigquery/#storage-write-api)。
  • useStorageWriteApiAtLeastOnce：使用 Storage Write API 時，指定寫入語意。如要使用至少一次語意 (https://beam.apache.org/documentation/io/built-in/google-bigquery/#at-least-once-semantics)，請將這個參數設為 true。如要使用僅一次語意，請將參數設為 false。只有在 useStorageWriteApi 為 true 時，這項參數才會生效。預設值為 false。
  • numStorageWriteApiStreams：使用 Storage Write API 時，指定寫入串流的數量。如果 useStorageWriteApi 為 true 且 useStorageWriteApiAtLeastOnce 為 false，則必須設定這個參數。預設為：0。
  • storageWriteApiTriggeringFrequencySec：使用 Storage Write API 時，指定觸發頻率 (以秒為單位)。如果 useStorageWriteApi 為 true 且 useStorageWriteApiAtLeastOnce 為 false，則必須設定這個參數。

執行範本

如要執行這個範本，請按照下列步驟操作：

  1. 在本機電腦上複製 DataflowTemplates 存放區。
  2. 切換至 v2/cdc-parent 目錄。
  3. 確認 Debezium 連接器已部署。
  4. 使用 Maven 執行 Dataflow 範本： 
    mvn exec:java -pl cdc-change-applier -Dexec.args="--runner=DataflowRunner \
    --inputSubscriptions=SUBSCRIPTIONS \
    --updateFrequencySecs=300 \
    --changeLogDataset=CHANGELOG_DATASET \
    --replicaDataset=REPLICA_DATASET \
    --project=PROJECT_ID \
    --region=REGION_NAME"

    更改下列內容：

    • PROJECT_ID：您要執行 Dataflow 工作的 Google Cloud 專案 ID
    • SUBSCRIPTIONS：您的 Pub/Sub 訂閱項目名稱逗號分隔清單
    • CHANGELOG_DATASET：用於變更記錄資料的 BigQuery 資料集
    • REPLICA_DATASET：用於副本資料表的 BigQuery 資料集

後續步驟