管理串流

Datastream API 可讓您透過 REST API 呼叫或 Google Cloud CLI，自動執行資料複製工作流程。您可以建立、驗證、修改和刪除串流，或擷取和列出串流和串流物件。

您也可以啟動、暫停、繼續或復原串流、管理物件補充、調整工作並行，以及串流大型 Oracle 物件。如需指令參考資料，請參閱 gcloud datastream streams。如需端對端逐步操作說明，請參閱「使用 Datastream API」。

建立串流

在本節中，您將瞭解如何建立串流，將資料從來源轉移至目的地。以下範例並非詳盡無遺，而是著重於 Datastream 的特定功能。如要處理特定用途，請搭配使用這些範例和 Datastream API 參考文件。

本節涵蓋下列用途：

範例 1：將特定物件串流至 BigQuery

在這個範例中，您會學到以下內容：

  • 從 MySQL 串流至 BigQuery
  • 在串流中加入一組物件
  • 將串流的寫入模式定義為「僅附加」
  • 補充串流中的所有物件

以下要求會從 schema1 提取所有資料表，並從 schema2 提取兩個特定資料表：tableA 和 tableC。事件會寫入 BigQuery 中的資料集。

要求不含 customerManagedEncryptionKey 參數，因此系統會使用 Google Cloud 內部金鑰管理系統 (而非 CMEK) 加密資料。

與執行歷來資料補充作業 (或快照) 相關聯的 backfillAll 參數會設為空白字典 ({})，這表示 Datastream 會從串流中包含的所有資料表補充歷來資料。

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=mysqlCdcStream
{
  "displayName": "MySQL CDC to BigQuery",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp",
    "mysqlSourceConfig": {
      "includeObjects": {
        "mysqlDatabases": [
          { "database": "schema1" },
          {
            "database": "schema2",
            "mysqlTables": [
              {
                "table": "tableA",
                "table": "tableC"
              }
            ]
          }
        ]
      },
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
          "location": "us",
          "datasetIdPrefix": "prefix_"
        }
      },
      "dataFreshness": "900s"
    }
  },
  "backfillAll": {}
}

gcloud

如要進一步瞭解如何使用 gcloud 建立串流，請參閱 Google Cloud SDK 說明文件。

範例 2：從 PostgreSQL 來源的串流中排除特定物件

在這個範例中，您會學到以下內容：

  • 從 PostgreSQL 串流至 BigQuery
  • 從串流中排除物件
  • 從補充作業中排除物件

下列程式碼顯示建立串流的要求，用於將資料從來源 PostgreSQL 資料庫移轉至 BigQuery。從來源 PostgreSQL 資料庫建立串流時，您必須在要求中指定兩個額外的 PostgreSQL 專屬欄位：

  • replicationSlot：必須先建立複製運算單元，才能設定 PostgreSQL 資料庫進行複製。您必須為每個串流建立一個複製槽。
  • publication：發布作業是一組資料表，您要從中複製變更。資料庫中必須有發布作業名稱，才能啟動串流。發布作業至少須包含串流 includeObjects 清單中指定的資料表。

與執行歷來資料補充 (或快照) 相關聯的 backfillAll 參數設為排除一個資料表。

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams?streamId=myPostgresStream
{
  "displayName": "PostgreSQL to BigQueryCloud Storage",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/connectionProfiles/postgresCp",
    "postgresqlSourceConfig": {
      "replicationSlot": "replicationSlot1",
      "publication": "publicationA",
      "includeObjects": {
        "postgresqlSchemas": {
          "schema": "schema1"
        }
      },
      "excludeObjects": {
        "postgresqlSchemas": [
          { "schema": "schema1",
        "postgresqlTables": [
          {
            "table": "tableA",
            "postgresqlColumns": [
              { "column": "column5" }
              ]
              }
            ]
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "dataFreshness": "900s",
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
           "location": "us",
           "datasetIdPrefix": "prefix_"
        }
      }
    }
  },
  "backfillAll": {
    "postgresqlExcludedObjects": {
        "postgresqlSchemas": [
          { "schema": "schema1",
            "postgresqlTables": [
              { "table": "tableA" }
            ]
          }
        ]
      }
    }
  }

gcloud

如要進一步瞭解如何使用 gcloud 建立串流，請參閱 Google Cloud SDK 說明文件。

範例 3：為串流指定僅附加寫入模式

串流至 BigQuery 時，您可以定義寫入模式：merge 或 appendOnly。詳情請參閱「設定寫入模式」。

如果在建立串流的要求中未指定寫入模式，系統會使用預設的 merge 模式。

下列要求說明如何在建立 MySQL 到 BigQuery 串流時，定義 appendOnly 模式。

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=appendOnlyStream
{
  "displayName": "My append-only stream",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp",
    "mysqlSourceConfig": {
      "includeObjects": {
        "mysqlDatabases": [
          { "database": "myMySqlDb"
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
          "location": "us",
          "datasetIdPrefix": "prefix_"
        }
      },
      "appendOnly": {}
    }
  },
  "backfillAll": {}
}

gcloud

如要進一步瞭解如何使用 gcloud 建立串流，請參閱 Google Cloud SDK 說明文件。

範例 4：串流至 BigQuery 中的其他專案

如果您在某個專案中建立 Datastream 資源，但想串流至 BigQuery 中的其他專案，可以使用類似下列的要求。

如果為目的地資料集指定 sourceHierarchyDatasets，則必須填入 projectId 欄位。

如果您為目的地資料集指定 singleTargetDataset，請以 projectId:datasetId 格式填入 datasetId 欄位。

REST

針對 sourceHierarchyDatasets：

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=crossProjectBqStream1
{
  "displayName": "My cross-project stream",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp",
    "mysqlSourceConfig": {
      "includeObjects": {
        "mysqlDatabases": [
          { "database": "myMySqlDb"
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
          "location": "us",
          "datasetIdPrefix": "prefix_"
        },
        "projectId": "myProjectId2"
      }
    }
  },
  "backfillAll": {}
}

針對 singleTargetDataset：

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=crossProjectBqStream2
{
  "displayName": "My cross-project stream",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp",
    "mysqlSourceConfig": {
      "includeObjects": {
        "mysqlDatabases": [
          { "database": "myMySqlDb"
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "singleTargetDataset": {
        "datasetId": "myProjectId2:myDatasetId"
      },
    }
  },
  "backfillAll": {}
}

gcloud

針對 sourceHierarchyDatasets：

  datastream streams create crossProjectBqStream1 --location=us-central1
  --display-name=my-cross-project-stream --source=source-cp --mysql-source-config=mysql_source_config.json
  --destination=destination-cp --bigquery-destination-config=source_hierarchy_cross_project_config.json
  --backfill-none

source_hierarchy_cross_project_config.json 設定檔的內容：

  {"sourceHierarchyDatasets": {"datasetTemplate": {"location": "us-central1", "datasetIdPrefix": "prefix_"}, "projectId": "myProjectId2"}}

針對 singleTargetDataset：

  datastream streams create crossProjectBqStream --location=us-central1
  --display-name=my-cross-project-stream --source=source-cp --mysql-source-config=mysql_source_config.json
  --destination=destination-cp --bigquery-destination-config=single_target_cross_project_config.json
  --backfill-none

single_target_cross_project_config.json 設定檔的內容：

  {"singleTargetDataset": {"datasetId": "myProjectId2:myDatastetId"}}

如要進一步瞭解如何使用 gcloud 建立串流，請參閱 Google Cloud SDK 說明文件。

範例 5：串流至 Cloud Storage 目的地

在這個範例中，您會學到以下內容：

  • 從 Oracle 串流至 Cloud Storage
  • 定義要納入串流的一組物件
  • 定義 CMEK，用於加密靜態資料

下列要求說明如何建立串流，將事件寫入 Cloud Storage 中的 bucket。

在這個範例要求中，事件會以 JSON 輸出格式寫入，且每 100 MB 或 30 秒就會建立新檔案 (覆寫 50 MB 和 60 秒的預設值)。

如果是 JSON 格式，您可以：

  • 在路徑中新增一致類型的結構定義檔案。因此，Datastream 會將兩個檔案寫入 Cloud Storage：JSON 資料檔案和 Avro 結構定義檔案。結構定義檔案的名稱與資料檔案相同，但副檔名為 .schema。

  • 啟用 gzip 壓縮功能，讓 Datastream 壓縮寫入 Cloud Storage 的檔案。

使用 backfillNone 參數後，要求會指定只將進行中的變更串流至目的地，不進行補充。

要求會指定客戶自行管理的加密金鑰參數，讓您控管用於加密 Google Cloud 專案中靜態資料的金鑰。這個參數是指 Datastream 用來加密資料的 CMEK，資料會從來源串流至目的地。並指定 CMEK 的金鑰環。

如要進一步瞭解金鑰環，請參閱「Cloud KMS 資源」。如要進一步瞭解如何使用加密金鑰保護資料，請參閱「Cloud Key Management Service (KMS)」。

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams?streamId=myOracleCdcStream
{
  "displayName": "Oracle CDC to Cloud Storage",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/
    connectionProfiles/OracleCp",
    "oracleSourceConfig": {
      "includeObjects": {
        "oracleSchemas": [
          {
            "schema": "schema1"
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "GcsBucketCp",
    "gcsDestinationConfig": {
      "path": "/folder1",
      "jsonFileFormat": {
        "schemaFileFormat": "AVRO_SCHEMA_FILE"
      },
      "fileRotationMb": 100,
      "fileRotationInterval": 30
    }
  },
  "customerManagedEncryptionKey": "projects/myProjectId1/locations/us-central1/
  keyRings/myRing/cryptoKeys/myEncryptionKey",
  "backfillNone": {}
}

gcloud

如要進一步瞭解如何使用 gcloud 建立串流，請參閱 Google Cloud SDK 說明文件。

範例 6：串流至 Apache Iceberg 資料表

在本範例中，您將瞭解如何設定串流，將資料從 MySQL 資料庫複製到 append-only 模式的 Apache Iceberg 資料表。建立要求前，請務必完成下列步驟：

  • 具備要用來儲存資料的 Cloud Storage bucket
  • 建立 Cloud 資源連結
  • 向 Cloud Storage bucket 授予 Cloud 資源連結存取權

接著，您可以使用下列要求建立串流：

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=mysqlIcebergStream
{
  "displayName": "MySQL to Apache Iceberg stream",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlIcebergCp",
    "mysqlSourceConfig": {
      "includeObjects": {
        "mysqlDatabases": [
          {
            "database": "my-mysql-database"
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "projects/myProjectId1/locations/us-central1/connectionProfiles/my-bq-cp-id",
    "bigqueryDestinationConfig": {
      "blmtConfig": {
        "bucket": "my-gcs-bucket-name",
        "rootPath": "my/folder",
        "connectionName": "my-project-id.us-central1.my-bigquery-connection-name",
        "fileFormat": "PARQUET",
        "tableFormat": "ICEBERG"
        },
      "singleTargetDataset": {
        "datasetId": "my-project-id:my-bigquery-dataset-id"
      },
      "appendOnly": {}
    }
  },
  "backfillAll": {}
}

gcloud

datastream streams create mysqlIcebergStream --location=us-central1
--display-name=mysql-to-bl-stream --source=source --mysql-source-config=mysql_source_config.json
--destination=destination --bigquery-destination-config=bl_config.json
--backfill-none

mysql_source_config.json 來源設定檔的內容：

{"excludeObjects": {}, "includeObjects": {"mysqlDatabases":[{"database":"my-mysql-database"}]}}

bl_config.json 設定檔的內容：

{ "blmtConfig": { "bucket": "my-gcs-bucket-name", "rootPath": "my/folder", "connectionName": "my-project-id.us-central1.my-bigquery-connection-name", "fileFormat": "PARQUET", "tableFormat": "ICEBERG" }, "singleTargetDataset": {"datasetId": "my-project-id:my-bigquery-dataset-id"}, "appendOnly": {} }

Terraform

resource "google_datastream_stream" "stream" {
  stream_id    = "mysqlBlStream"
  location     = "us-central1"
  display_name = "MySQL to Apache Iceberg stream"

  source_config {
    source_connection_profile = "/projects/myProjectId1/locations/us-central1/streams/mysqlBlCp"
    mysql_source_config {
      include_objects {
        mysql_databases {
          database = "my-mysql-database"
        }
      }
    }
  }

  destination_config {
    destination_connection_profile = "projects/myProjectId1/locations/us-central1/connectionProfiles/my-bq-cp-id"
    bigquery_destination_config {
      single_target_dataset {
        dataset_id = "my-project-id:my-bigquery-dataset-id"
      }
      blmt_config {
        bucket          = "my-gcs-bucket-name"
        table_format    = "ICEBERG"
        file_format     = "PARQUET"
        connection_name = "my-project-id.us-central1.my-bigquery-connection-name"
        root_path       = "my/folder"
      }
      append_only {}
    }
  }

  backfill_none {}
}

範例 7：使用嚴格 JSON 模式從 MongoDB 串流

在這個範例中，您將設定串流，使用嚴格 JSON 模式將 MongoDB 來源中的資料複製到 BigQuery。

根據預設，Datastream 會使用標準 JSON 模式 (CANONICAL)，優先確保資料保真度。如果下游應用程式需要與標準 JSON 剖析器相容，您可以指定嚴格 JSON 模式 (STRICT)。

從 MongoDB 來源建立串流時：

  • API 中 jsonMode 的有效值為 CANONICAL (預設) 和 STRICT。
  • 在 gcloud CLI 中，JSON 模式沒有獨立旗標。請改為透過傳遞至 --mongodb-source-config 旗標的設定檔，設定 MongoDB 來源設定。

REST

在 Datastream REST API 中，將 sourceConfig.mongodbSourceConfig 物件中的 jsonMode 設為 "STRICT"：

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/streams?streamId=STREAM_ID
{
  "displayName": "STREAM_DISPLAY_NAME",
  "sourceConfig": {
    "sourceConnectionProfile": "projects/PROJECT_ID/locations/LOCATION/connectionProfiles/SOURCE_PROFILE",
    "mongodbSourceConfig": {
      "jsonMode": "STRICT",
      "includeObjects": {
        "databases": [
          {
            "database": "DATABASE_NAME",
            "collections": [
              {
                "collection": "COLLECTION_NAME"
              }
            ]
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfile": "projects/PROJECT_ID/locations/LOCATION/connectionProfiles/DEST_PROFILE",
    "bigqueryDestinationConfig": {
      "singleTargetDataset": {
        "datasetId": "PROJECT_ID:DATASET_ID"
      }
    }
  }
}

更改下列內容：

  • PROJECT_ID： Google Cloud 專案的 ID。
  • LOCATION：建立串流的區域，例如 us-central1。
  • STREAM_ID：串流的 ID。
  • STREAM_DISPLAY_NAME：容易使用的串流名稱。
  • SOURCE_PROFILE：來源 MongoDB 連線設定檔的 ID。
  • DEST_PROFILE：目的地連線設定檔的 ID。
  • DATABASE_NAME：來源 MongoDB 叢集中的資料庫名稱。
  • COLLECTION_NAME：資料庫中的集合名稱。
  • DATASET_ID：BigQuery 中目的地資料集的 ID。

gcloud

如要使用 gcloud CLI 建立含有 MongoDB 來源和嚴格 JSON 模式的串流，請按照下列步驟操作：

  1. 以 JSON 或 YAML 格式建立來源設定檔，指定 "jsonMode": "STRICT" 和要複製的物件。

JSON 設定檔，例如 mongo_source_config.json：

{
  "jsonMode": "STRICT",
  "includeObjects": {
    "databases": [
      {
        "database": "DATABASE_NAME",
        "collections": [
          {
            "collection": "COLLECTION_NAME"
          }
        ]
      }
    ]
  }
}

YAML 設定檔，例如 mongo_source_config.yaml：

jsonMode: STRICT
includeObjects:
  databases:
    - database: DATABASE_NAME
      collections:
        - collection: COLLECTION_NAME
  1. 建立串流時，請將設定檔傳遞至 --mongodb-source-config 旗標：
gcloud datastream streams create STREAM_ID \
    --location=LOCATION \
    --display-name="STREAM_DISPLAY_NAME" \
    --source=projects/PROJECT_ID/locations/LOCATION/connectionProfiles/SOURCE_PROFILE \
    --destination=projects/PROJECT_ID/locations/LOCATION/connectionProfiles/DEST_PROFILE \
    --mongodb-source-config=mongo_source_config.json \
    --bigquery-destination-config=bigquery_destination_config.json

bigquery_destination_config.json 設定檔的內容：

{
  "singleTargetDataset": {
    "datasetId": "PROJECT_ID:DATASET_ID"
  }
}

更改下列內容：

  • STREAM_ID：串流的 ID。
  • LOCATION：建立串流的區域，例如 us-central1。
  • STREAM_DISPLAY_NAME：容易使用的串流名稱。
  • PROJECT_ID： Google Cloud 專案的 ID。
  • SOURCE_PROFILE：來源 MongoDB 連線設定檔的 ID。
  • DEST_PROFILE：目的地連線設定檔的 ID。
  • DATABASE_NAME：來源 MongoDB 叢集中的資料庫名稱。
  • COLLECTION_NAME：資料庫中的集合名稱。
  • DATASET_ID：BigQuery 中目的地資料集的 ID。

如要進一步瞭解如何使用 gcloud 建立串流，請參閱 Google Cloud SDK 說明文件。

驗證串流定義

建立串流前，您可以先驗證其定義。這樣一來，您就能確保所有驗證檢查都通過，且串流建立後會成功執行。

驗證串流時，系統會檢查：

  • 來源是否已正確設定，可讓 Datastream 從中串流資料。
  • 串流是否能同時連線至來源和目的地。
  • 串流的端對端設定。

如要驗證串流，請在要求主體前的網址中加入 &validate_only=true：

POST "https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/streams?streamId=STREAM_ID&validate_only=true"

提出這項要求後，您會看到 Datastream 對來源和目的地執行的驗證檢查，以及檢查是否通過。如果驗證檢查未通過，系統會顯示失敗原因和修正問題的方法。

舉例來說，假設您有客戶自行管理的加密金鑰 (CMEK)，並希望 Datastream 使用該金鑰，加密從來源串流至目的地的資料。驗證串流時，Datastream 會確認金鑰是否存在，以及 Datastream 是否有權使用金鑰。如果不符合上述任一條件，驗證串流時會傳回下列錯誤訊息：

CMEK_DOES_NOT_EXIST_OR_MISSING_PERMISSIONS

如要解決這個問題，請確認您提供的金鑰存在，且 Datastream 服務帳戶具備該金鑰的 cloudkms.cryptoKeys.get 權限。

完成適當修正後，請再次提出要求，確保所有驗證檢查都通過。以上述範例來說，CMEK_VALIDATE_PERMISSIONS 檢查不會再傳回錯誤訊息，但狀態會是 PASSED。

取得串流相關資訊

下列程式碼顯示擷取串流相關資訊的要求。這類資訊包括：

  • 串流名稱 (專屬 ID)
  • 容易使用的串流名稱 (顯示名稱)
  • 串流的建立時間和上次更新時間
  • 與串流相關聯的來源和目的地連線設定檔資訊
  • 串流狀態

REST

GET https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/streams/STREAM_ID

回應如下：

{
  "name": "myOracleCdcStream",
  "displayName": "Oracle CDC to Cloud Storage",
  "createTime": "2019-12-15T15:01:23.045123456Z",
  "updateTime": "2019-12-15T15:01:23.045123456Z",
  "sourceConfig": {
    "sourceConnectionProfileName": "myOracleDb",
    "oracleSourceConfig": {
      "includeObjects": {
        "oracleSchemas": [
          {
            "schema": "schema1"
          },
          {
            "schema": "schema3",
            "oracleTables": [
              { "table": "tableA" },
              { "table": "tableC" }
            ]
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "myGcsBucket",
    "gcsDestinationConfig": {
      "path": "/folder1",
      "avroFileFormat": {},
      "fileRotationMb": 100,
      "fileRotationInterval": 60
    }
  },
  "state": "RUNNING"
  "customerManagedEncryptionKey": "projects/myProjectId1/locations/us-central1/
  keyRings/myRing/cryptoKeys/myEncryptionKey",
  "backfillAll": {}
}

gcloud

如要進一步瞭解如何使用 gcloud 擷取串流相關資訊，請參閱 Google Cloud SDK 說明文件。

列出串流

下列程式碼顯示要求，用於擷取指定專案和位置中所有串流的清單。

REST

GET https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams

gcloud

如要進一步瞭解如何使用 gcloud 擷取所有串流的資訊，請參閱 Google Cloud SDK 說明文件。

列出串流的物件

下列程式碼顯示擷取串流所有物件相關資訊的要求。

REST

GET https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID/objects

gcloud

如要進一步瞭解如何使用 gcloud 擷取串流中所有物件的資訊，請參閱 Google Cloud SDK 說明文件。

傳回的物件清單可能如下所示：

REST

{
  "streamObjects": [
    {
      "name": "projects/myProjectId1/locations/us-central1/streams/myStream/
      objects/object1",
      "displayName": "employees.salaries",
      "backfillJob": {
        "state": "ACTIVE",
        "trigger": "AUTOMATIC",
        "lastStartTime": "2021-10-18T12:12:26.344878Z"
      },
      "sourceObject": {
        "mysqlIdentifier": {
          "database": "employees",
          "table": "salaries"
        }
      }
    },
    {
      "name": "projects/myProjectId1/locations/us-central1/streams/myStream/
      objects/object2",
      "displayName": "contractors.hours",
      "sourceObject": {
        "mysqlIdentifier": {
          "database": "contractors",
          "table": "hours"
        }
      }
    },
    {
      "name": "projects/myProjectId1/locations/us-central1/streams/myStream/
      objects/object3",
      "displayName": "employees.departments",
      "backfillJob": {
        "state": "COMPLETED",
        "trigger": "AUTOMATIC",
        "lastStartTime": "2021-10-18T11:26:12.869880Z",
        "lastEndTime": "2021-10-18T11:26:28.405653Z"
      },
      "sourceObject": {
        "mysqlIdentifier": {
          "database": "employees",
          "table": "departments"
        }
      }
    }
  ]
}

gcloud

如要進一步瞭解如何使用 gcloud 列出串流物件，請參閱 Google Cloud SDK 說明文件。

啟動串流

下列程式碼顯示啟動串流的要求。

在要求中使用 updateMask 參數，要求主體就只需包含您指定的欄位。如要啟動串流，請將 state 欄位的值從 CREATED 變更為 RUNNING。

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=state
{
  "state": "RUNNING"
}

gcloud

如要進一步瞭解如何使用 gcloud 啟動串流，請參閱 Google Cloud SDK 說明文件。

暫停串流

下列程式碼顯示暫停執行中串流的要求。

在本例中，為 updateMask 參數指定的欄位是 state 欄位。暫停串流後，狀態會從 RUNNING 變更為 PAUSED。

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=state
{
  "state": "PAUSED"
}

gcloud

如要進一步瞭解如何使用 gcloud 暫停串流，請參閱 Google Cloud SDK 說明文件。

繼續執行串流

下列程式碼顯示恢復暫停串流的要求。

在本例中，為 updateMask 參數指定的欄位是 state 欄位。恢復串流後，狀態會從「PAUSED」變更為「RUNNING」。

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=state
{
  "state": "RUNNING"
}

gcloud

如要進一步瞭解如何使用 gcloud 繼續串流，請參閱 Google Cloud SDK 說明文件。

復原串流

您可以使用 RunStream 方法，復原永久失敗的串流。每種來源資料庫類型都有各自的串流復原作業定義。詳情請參閱「復原串流」。

復原 MySQL 或 Oracle 來源的串流

下列程式碼範例顯示從各種記錄檔位置復原 MySQL 或 Oracle 來源串流的要求：

REST

從目前位置復原串流。這是預設選項：

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run

從下一個可用位置復原串流：

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "nextAvailableStartPosition": {}
  }
}

從最新位置復原串流：

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "mostRecentStartPosition": {}
  }
}

從特定位置復原串流 (MySQL 二進位記錄檔複製)：

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlLogPosition": {
        "logFile": "NAME_OF_THE_LOG_FILE"
        "logPosition": POSITION
      }
    }
  }
}

更改下列內容：

  • NAME_OF_THE_LOG_FILE：要復原串流的記錄檔名稱
  • POSITION：記錄檔中要復原串流的位置。如未提供值，Datastream 會從檔案開頭復原串流。

例如：

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlLogPosition": {
        "logFile": "binlog.001"
        "logPosition": 4
      }
    }
  }
}

從特定位置復原串流 (MySQL GTID 複製)：

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlGtidPosition": {
        "gtidSet": "GTID_SET"
      }
    }
  }
}

將 GTID_SET 替換為一或多個 GTID 或 GTID 範圍，以便從中復原串流。

例如：

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlGtidPosition": {
        "gtidSet": "22cc56f5-3862-379a-9af5-051c59baef9d:1-561143685:561143688-591036613,b8d7df02-832b-32b9-bec7-2018806b76f6:1-3"
      }
    }
  }
}

從特定位置復原串流 (Oracle)：

POST https://datastream.googleapis.com/v1/projects/[project-id]/locations/
[location]/streams/[stream-id]:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "oracleScnPosition": {
        "scn": scn
      }
    }
  }
}
 請將 scn 替換為重做記錄檔中的系統變更編號 (SCN)，以便從該檔案復原串流。這是必填欄位。

例如：

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "oracleScnPosition": {
        "scn": 234234
      }
    }
  }
}

如要進一步瞭解可用的復原選項，請參閱「復原串流」。

gcloud

不支援使用 gcloud 復原串流。

復原 PostgreSQL 來源的串流

下列程式碼範例顯示復原 PostgreSQL 來源串流的要求。復原期間，串流會從為串流設定的複製運算單元中，第一個記錄檔序號 (LSN) 開始讀取。

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "nextAvailableStartPosition": {}
  }
}

例如：

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run

如要變更複製運算單元，請先使用新的複製運算單元名稱更新串流：

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=sourceConfig.postgresqlSourceConfig.replicationSlot
{
  "sourceConfig": {
    "postgresqlSourceConfig": {
      "replicationSlot": "NEW_REPLICATION_SLOT_NAME"
    }
  }
}

gcloud

不支援使用 gcloud 復原串流。

復原 SQL Server 來源的串流

下列程式碼範例顯示復原 SQL Server 來源串流的要求範例。

REST

從第一個可用位置復原串流：

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run

例如：

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/mySqlServerStreamId:run

從偏好的記錄序號復原串流：

POST https://datastream.googleapis.com/v1/projects/[project-id]/locations/
[location]/streams/[stream-id]:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "sqlServerLsnPosition": {
        "lsn": lsn
      }
    }
  }
}

例如：

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/mySqlServerStreamId:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "sqlServerLsnPosition": {
        "lsn": 0000123C:0000BA78:0004
      }
    }
  }
}

gcloud

不支援使用 gcloud 復原串流。

從特定位置開始或繼續播放串流

您可以從特定位置啟動串流，或繼續執行暫停的串流，適用於 MySQL、Oracle 和 SQL Server 來源。如要使用外部工具執行補充作業，或從指定位置啟動 CDC，這個做法就很實用：

  • 如果是 MySQL 來源，您必須指定二進位記錄位置或 GTID 集。
  • 如果是 Oracle 來源，您必須在重做記錄檔中指定系統變更編號 (SCN)。
  • 如果是 SQL Server 來源，您必須在變更資料表中指定記錄檔序號 (LSN)。

下列程式碼範例說明如何要求從特定位置開始或繼續播放已建立的串流。

從特定二進位記錄位置啟動或繼續串流 (MySQL)：

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlLogPosition": {
        "logFile": "NAME_OF_THE_LOG_FILE"
        "logPosition": POSITION
      }
    }
  }
}

更改下列內容：

  • NAME_OF_THE_LOG_FILE：要開始串流的記錄檔名稱。
  • POSITION：記錄檔中要開始串流的位置。如未提供值，Datastream 會從檔案開頭開始讀取。

例如：

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlLogPosition": {
        "logFile": "binlog.001"
        "logPosition": 2
      }
    }
  }
}

gcloud

不支援使用 gcloud 從特定位置開始或繼續播放串流。如要瞭解如何使用 gcloud 啟動或繼續串流，請參閱 Cloud SDK 說明文件。

從特定 GTID 集 (MySQL) 啟動或繼續串流：

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlGtidPosition": {
        "gtidSet": "GTID_SET"
      }
    }
  }
}

將 GTID_SET 替換為一或多個單一 GTID 或 GTID 範圍，做為串流的起始或恢復點。

例如：

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlGtidPosition": {
        "gtidSet": "22cc56f5-3862-379a-9af5-051c59baef9d:1-561143685:561143688-591036613,b8d7df02-832b-32b9-bec7-2018806b76f6:3-7"
      }
    }
  }
}

gcloud

不支援使用 gcloud 從特定位置開始或繼續播放串流。如要瞭解如何使用 gcloud 啟動或繼續串流，請參閱 Cloud SDK 說明文件。

從重做日誌檔 (Oracle) 中的特定系統變更編號，啟動或繼續串流：

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "oracleScnPosition": {
        "scn": scn
      }
    }
  }
}
 將 scn 替換成重做記錄檔中的系統變更編號 (SCN)，也就是要開始串流的位置。這是必填欄位。

例如：

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "oracleScnPosition": {
        "scn": 123123
      }
    }
  }
}

gcloud

不支援使用 gcloud 從特定位置開始或繼續播放串流。如要瞭解如何使用 gcloud 啟動串流，請參閱 Cloud SDK 說明文件。

從特定記錄序號 (SQL Server) 啟動或繼續串流：

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "sqlServerLsnPosition": {
        "lsn": "lsn"
      }
    }
  }
}

將 lsn 替換為變更資料表中要開始串流的記錄序號。這是必填欄位。

例如：

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "sqlServerLsnPosition": {
        "lsn": "0000123C0000BA780004"
      }
    }
  }
}

詳情請參閱 Datastream API 參考文件。

gcloud

不支援使用 gcloud 從特定位置開始或繼續播放串流。如要瞭解如何使用 gcloud 啟動串流，請參閱 Cloud SDK 說明文件。

修改串流

下列程式碼顯示更新串流檔案輪替設定的要求，每 75 MB 或 45 秒輪替一次檔案。

在本例中，為 updateMask 參數指定的欄位包括 fileRotationMb 和 fileRotationInterval 欄位，分別以 destinationConfig.gcsDestinationConfig.fileRotationMb 和 destinationConfig.gcsDestinationConfig.fileRotationInterval 旗標表示。

REST

PATCH https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams/myStream?updateMask=destinationConfig.gcsDestinationConfig.
fileRotationMb,destinationConfig.gcsDestinationConfig.fileRotationInterval
{
  "destinationConfig": {
    "gcsDestinationConfig": {
      "fileRotationMb": 75,
      "fileRotationInterval": 45
    }
  }
}

下列程式碼顯示一項要求，在 Datastream 寫入 Cloud Storage 的檔案路徑中，加入一致類型的結構定義檔案。因此，Datastream 會寫入兩個檔案：JSON 資料檔案和 Avro 結構定義檔案。

在本範例中，指定的欄位是 jsonFileFormat 欄位，由 destinationConfig.gcsDestinationConfig.jsonFileFormat 旗標表示。

PATCH https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams/myStream?updateMask=destinationConfig.gcsDestinationConfig.
jsonFileFormat
{
  "destinationConfig": {
    "gcsDestinationConfig": {
      "jsonFileFormat" {
        "schemaFileFormat": "AVRO_SCHEMA_FILE"
      }  
    }
  }
}

下列程式碼顯示 Datastream 要求，除了資料的持續變更外，還要將現有資料從來源資料庫複製到目的地。

程式碼的 oracleExcludedObjects 部分會顯示無法回填至目的地的資料表和結構定義。

在這個範例中，除了 schema3 中的 tableA 之外，所有資料表和結構定義都會補充。

PATCH https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams/myStream?updateMask=backfillAll
{
  "backfillAll": {
    "oracleExcludedObjects": {
      "oracleSchemas": [
        {
          "schema": "schema3",
          "oracleTables": [
            { "table": "tableA" }
          ]
        }
      ]
    }
  }
}

gcloud

如要進一步瞭解如何使用 gcloud 修改串流，請參閱 Google Cloud SDK 說明文件。

啟動串流物件的補充作業

Datastream 中的串流可以補充歷來資料，並將持續變更串流至目的地。持續變更一律會從來源串流至目的地。不過，您可以指定是否要串流歷來資料。

如要將歷來資料從來源串流至目的地，請使用 backfillAll 參數。

Datastream 也可讓您只串流特定資料庫資料表的歷來資料。如要這麼做，請使用 backfillAll 參數，並排除不需歷來資料的資料表。

如果只想將持續變更串流至目的地，請使用 backfillNone 參數。如果之後想讓 Datastream 將所有現有資料的快照從來源串流至目的地，則必須手動為包含這些資料的物件啟動補充作業。

如果來源和目的地之間的資料不同步，也需要對物件啟動補充作業。舉例來說，使用者可能會不小心刪除目的地的資料，導致資料遺失。在這種情況下，啟動物件補充作業可做為「重設機制」，因為所有資料都會一次串流至目的地。因此，資料會在來源和目的地之間同步。

您必須先擷取串流物件的相關資訊，才能啟動該物件的補充作業。

每個物件都有一個 OBJECT_ID，用於識別物件。您可以使用 OBJECT_ID 啟動串流的回填作業。

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID/objects/OBJECT_ID:startBackfillJob

gcloud

如要進一步瞭解如何使用 gcloud 為串流物件啟動補充作業，請參閱 Google Cloud SDK 說明文件。

停止補充串流物件

啟動串流物件的補充作業後，您可以停止該物件的補充作業。舉例來說，如果使用者修改資料庫結構定義，結構定義或資料可能會損毀。您不希望將這個結構定義或資料串流至目的地，因此停止回填物件。

您也可以停止物件補充作業，以達到負載平衡的目的。Datastream 可同時執行多項補充作業。這可能會增加來源的負載。如果負載量過大，請停止所有物件的補充作業，然後逐一啟動補充作業。

如要停止補充串流物件，請先要求擷取串流所有物件的相關資訊。每個傳回的物件都有 OBJECT_ID，可專門識別該物件。您可以使用 OBJECT_ID 停止串流的回填作業。

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID/objects/OBJECT_ID:stopBackfillJob

gcloud

如要進一步瞭解如何使用 gcloud 停止串流物件的回填作業，請參閱 Google Cloud SDK 說明文件。

變更 CDC 並行任務數量上限

下列程式碼說明如何將 MySQL 串流的變更資料擷取 (CDC) 工作並行數量上限設為 7。

在本例中，為 updateMask 參數指定的欄位是 maxConcurrentCdcTasks 欄位。將這個值設為 7，即可將並行 CDC 工作數量上限從先前的值變更為 7。可使用的值介於 0 到 50 (含) 之間。如未定義值或定義為 0，系統會將串流設為預設值 5。

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=sourceConfig.mysqlSourceConfig.maxConcurrentCdcTasks
{
  "sourceConfig": {
    "mysqlSourceConfig": {
      "maxConcurrentCdcTasks": "7"
      }
    }  
}

gcloud

如要進一步瞭解如何使用 gcloud，請參閱 Google Cloud SDK 說明文件。

變更並行補充作業數量上限

下列程式碼說明如何將 MySQL 串流的並行補充工作數量上限設為 25。

在本例中，為 updateMask 參數指定的欄位是 maxConcurrentBackfillTasks 欄位。將值設為 25，即可將並行補充任務數量上限從先前的值變更為 25。可使用的值介於 0 到 50 (含) 之間。如未定義值或定義為 0，系統會將串流設為預設值 16。

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/
streams/STREAM_ID?updateMask=sourceConfig.mysqlSourceConfig.maxConcurrentBackfillTasks
{
  "sourceConfig": {
    "mysqlSourceConfig": {
      "maxConcurrentBackfillTasks": "25"
      }
    }  
}

gcloud

如要進一步瞭解如何使用 gcloud，請參閱 Google Cloud SDK 說明文件。

啟用 Oracle 來源的大型物件串流功能

您可以為 Oracle 來源的串流啟用大型物件串流，例如二進位大型物件 (BLOB)、字元大型物件 (CLOB) 和國家字元大型物件 (NCLOB)。streamLargeObjects 旗標可讓您在新的和現有串流中加入大型物件。這個旗標是在串流層級設定，因此您不必指定大型物件資料類型的資料欄。

以下範例說明如何建立串流，以便串流大型物件。

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams?streamId=myOracleLobStream
{
  "displayName": "Oracle LOB stream",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/connectionProfiles/OracleCp",
    "oracleSourceConfig": {
      "includeObjects": {
        "oracleSchemas": [
          {
            "schema": "schema1",
            "oracleTables": [
              {
                "table": "tableA",
                "oracleColumns": [
                  {
                    "column": "column1,column2"
                  }
                ]
              }
            ]
          }
        ]
      },
      "excludeObjects": {},
      "streamLargeObjects": {}
    }
  }
}

gcloud

如要進一步瞭解如何使用 gcloud 更新串流，請參閱 Google Cloud SDK 說明文件。

刪除串流

下列程式碼顯示刪除串流的要求。

REST

DELETE https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID

gcloud

如要進一步瞭解如何使用 gcloud 刪除串流，請參閱 Google Cloud SDK 說明文件。

後續步驟