活動與串流

Datastream 會將複製的資料整理成三層階層，分別是串流、物件和事件。串流會將資料來源連結至目的地，而物件則代表該串流中的特定來源實體，例如資料庫資料表。「事件」會擷取物件的單一資料變更，例如資料庫插入。

每個事件都包含資料和中繼資料，可識別來源串流和物件。您可以使用這項中繼資料追蹤變更來源、移除目的地中的重複記錄，以及維持下游消費者的排序。

關於事件

每個事件都包含三種資料：

  • 事件資料：代表資料本身的變更，來自串流來源的物件。每個事件都包含已變更的整列資料。
  • 一般中繼資料：這類中繼資料會顯示在 Datastream 產生的每個事件中，用於執行移除目的地重複資料等動作。
  • 特定來源的中繼資料：這類中繼資料會顯示在特定串流來源產生的每個事件中。這類中繼資料會因來源而異。

事件資料

事件資料是來自串流來源的指定物件每次變更的酬載。

事件格式為 Avro 或 JSON。

使用 Avro 格式時，事件會包含每個資料欄的索引和值。使用資料欄索引，即可從 Avro 標頭的結構定義中擷取資料欄名稱和一致類型。

使用 JSON 格式時，事件會包含每個資料欄的名稱和值。

事件中繼資料可用於收集事件來源的相關資訊，以及移除目的地中的重複資料，並依下游消費者排序事件。

下表列出並說明一般和來源專屬事件中繼資料的欄位和資料類型。

一般中繼資料

這項中繼資料在所有類型的串流中都一致。

欄位 Avro 類型 JSON 類型 說明
stream_name 字串 字串 建立時定義的不重複串流名稱。
read_method 字串 字串

指出資料是否使用變更資料擷取 (CDC) 方法從來源讀取，做為歷來資料補充作業的一部分，或是在 CDC 複製期間復原交易時建立的補充工作。

可能的值包括：

  • cdc
  • backfill
  • oracle-cdc-logminer
  • oracle-backfill
  • oracle-supplementation
  • mysql-cdc-binlog
  • mysql-backfill-incremental
  • mysql-backfill-fulldump
  • postgres-cdc-wal
  • postgresql-backfill
  • sqlserver-cdc
  • sqlserver-backfill
  • salesforce-cdc
  • salesforce-backfill
object 字串 字串 用於將不同類型事件分組的名稱，通常是來源中的資料表或物件名稱。
schema_key 字串 字串 事件統一結構定義的專屬 ID。
uuid 字串 字串 Datastream 產生的事件專屬 ID。
read_timestamp timestamp-millis 字串 Datastream 讀取記錄的時間戳記 (世界標準時間)，以毫秒為單位。
source_timestamp timestamp-millis 字串 來源記錄變更的時間戳記 (世界標準時間)，以毫秒為單位。
sort_keys {"type": "array", "items": ["string", "long"]} 陣列 可用來依事件發生順序排序事件的值陣列。

來源專屬中繼資料

這項中繼資料與來源資料庫的 CDC 和補充事件相關聯。如要查看這項中繼資料，請從下方的下拉式選單中選取來源。

來源 欄位 Avro 類型 JSON 類型 說明
MySQL log_file 字串 字串 使用二進位記錄檔複製功能時，Datastream 在 CDC 複製作業中提取事件的記錄檔。
MySQL log_position long long 使用二進位記錄式複製功能時，MySQL 二進位記錄中的記錄位置 (偏移)。
MySQL binlog_file 字串 字串 使用以 GTID 為基礎的複製模式時，Datastream 在 CDC 複製作業中提取事件的二進位記錄檔。
MySQL binlog_position long long 使用以 GTID 為基礎的複製作業時，MySQL 二進位記錄中的位置 (偏移)。
MySQL gtid 字串 字串 使用以 GTID 為基礎的複製功能時，用於識別事件交易的全域交易 ID (GTID)。
MySQL datastream_master_server_uuid 字串 字串 使用以 GTID 為依據的複製作業時，Datastream 會從這個 UUID 讀取事件，該 UUID 屬於 MySQL 來源伺服器執行個體。
MySQL datastream_master_server_id long long 使用以 GTID 為依據的複製作業時，Datastream 會從這個 MySQL 來源伺服器執行個體讀取事件。
MySQL primary_keys 字串陣列 字串陣列 構成資料表主鍵的一或多個資料欄名稱清單。如果資料表沒有主鍵，這個欄位會留空。
MySQL is_deleted 布林值 布林值
  • true 值表示資料列已在來源中刪除。
  • false 值表示資料列未刪除。
MySQL database 字串 字串 與事件相關聯的資料庫。
MySQL table 字串 字串 與事件相關聯的資料表。
MySQL change_type 字串 字串

事件代表的變更類型 (INSERT、UPDATE-INSERT、UPDATE-DELETE 和 DELETE)。
Oracle log_file 字串 字串 Datastream 在 CDC 複製作業中，從中提取事件的記錄檔。
Oracle scn long long Oracle 交易記錄中的記錄位置 (偏移)。
Oracle row_id 字串 字串 Oracle 的 row_id。
Oracle is_deleted 布林值 布林值
  • true 值表示資料列已在來源中刪除。
  • false 值表示資料列未刪除。
Oracle database 字串 字串 與事件相關聯的資料庫。
Oracle schema 字串 字串 與事件資料表相關聯的結構定義。
Oracle table 字串 字串 與事件相關聯的資料表。
Oracle change_type 字串 字串

事件代表的變更類型 (INSERT、UPDATE-INSERT、UPDATE-DELETE 和 DELETE)。
Oracle tx_id 字串 字串 事件所屬的交易 ID。
Oracle rs_id 字串 字串 記錄集 ID。rs_id 和 ssn 的組合可唯一識別 V$LOGMNR_CONTENTS 中的資料列。rs_id 可明確識別產生資料列的重做記錄。
Oracle ssn long long SQL 序號。這個數字會與 rs_id 搭配使用，用於識別 V$LOGMNR_CONTENTS 中的資料列。
PostgreSQL schema 字串 字串 與事件資料表相關聯的結構定義。
PostgreSQL table 字串 字串 與事件相關聯的資料表。
PostgreSQL is_deleted 布林值 布林值
  • true 值表示資料列已在來源中刪除。
  • false 值表示資料列未刪除。
PostgreSQL change_type 字串 字串 事件代表的變更類型 (INSERT、UPDATE、DELETE)。
PostgreSQL tx_id 字串 字串 事件所屬的交易 ID。
PostgreSQL lsn 字串 字串 目前項目的記錄序號。
PostgreSQL primary_keys 字串陣列 字串陣列 構成資料表主鍵的一或多個資料欄名稱清單。如果資料表沒有主鍵，這個欄位會留空。
SQL Server table 字串 字串 與事件相關聯的資料表。
SQL Server database long long 與事件相關聯的資料庫。
SQL Server schema 字串陣列 字串陣列 與事件資料表相關聯的結構定義。
SQL Server is_deleted 布林值 布林值
  • true 值表示資料列已在來源中刪除。
  • false 值表示資料列未刪除。
SQL Server lsn 字串 字串 事件的記錄序號。
SQL Server tx_id 字串 字串 事件所屬的交易 ID。
SQL Server physical_location 整數陣列 整數陣列 記錄檔記錄的實體位置，由三個整數描述：記錄的檔案 ID、頁面 ID 和槽 ID。
SQL Server replication_index 字串陣列 字串陣列 索引的資料欄名稱清單，可不重複識別資料表中的資料列。
SQL Server change_type 字串 字串

事件代表的變更類型 (INSERT、UPDATE、DELETE)。
Spanner commit_timestamp timestamp-micros 字串 事件的修訂時間戳記。
Spanner snapshot 布林值 布林值 事件是否為補充快照事件。
Spanner project_id 字串 字串 Spanner 專案 ID。
Spanner instance_id 字串 字串 Spanner 執行個體 ID。
Spanner database_id 字串 字串 Spanner 資料庫 ID。
Spanner schema 字串 字串 與事件資料表相關聯的結構定義。如果資料表不在具名結構定義中，則為空白。
Spanner table 字串 字串 Spanner 資料表。
Spanner change_stream_name 字串 字串 Spanner 變更串流。
Spanner server_transaction_id 字串 字串 與事件相關聯的交易 ID。
Spanner record_sequence long 或 string long 或 string 與事件相關聯的記錄序列。如果是以 IMMUTABLE_KEY_RANGE 分區模式建立的變更串流，這個值會是 long 類型的數字序號。如果是以 MUTABLE_KEY_RANGE 分區模式建立的變更串流，這個值會是資料變更記錄中的 record_sequence 字串，例如 1cac8f3f38eeaf53-00000000。
Spanner mod_number long long 與事件相關聯的修改編號。
Spanner transaction_tag 字串 字串 相關交易的標記。
Spanner number_of_records_in_transaction long long 相關交易中的記錄數量。
Spanner value_capture_type 字串 字串 變更串流的值擷取類型。可能的值為 NEW_ROW 和 NEW_ROW_AND_OLD_VALUES。
Spanner change_type 字串 字串 事件代表的變更類型 (INSERT、UPDATE 或 DELETE)。
Spanner primary_keys 字串陣列 字串陣列 資料表的主鍵欄清單。
Spanner is_deleted 布林值 布林值 變更是否代表 DELETE 事件。
Salesforce object_name 字串 字串

與事件相關聯的 Salesforce 物件名稱。
Salesforce domain 字串 字串

與事件相關聯的網域名稱。
Salesforce is_deleted 布林值 布林值
  • true 值表示資料列已在來源中刪除。
  • false 值表示資料列未刪除。
Salesforce change_type 字串 字串

事件代表的變更類型 (INSERT、UPDATE、DELETE)。
Salesforce primary_keys 字串陣列 字串陣列 組成資料表主鍵的資料欄名稱清單。如果資料表沒有主鍵，這個欄位會留空。
MongoDB database 字串 字串 與事件相關聯的資料庫。
MongoDB collection 字串 字串 與事件相關聯的集合。集合類似於關聯式資料庫中的資料表。
MongoDB change_type 字串 字串 事件代表的變更類型 (CREATE、UPDATE 和 DELETE)。
MongoDB is_deleted 布林值 布林值
  • true 值表示資料列已在來源中刪除。
  • false 值表示資料列未刪除。
MongoDB primary_keys 字串陣列 字串陣列 _id 欄位，做為集合中每份文件的主鍵。

事件流程範例

這個流程說明來源資料庫中 SAMPLE 資料表單一資料列上，三個連續作業產生的事件：INSERT、UPDATE 和 DELETE。

時間 THIS_IS_MY_PK (int) FIELD1 (nchar 可為空值) FIELD2 (nchar non-null)>
0 1231535353 foo TLV
1 1231535353 空值 TLV

INSERT (T0)

訊息酬載包含整個新資料列。

{
  "stream_name": "projects/myProj/locations/myLoc/streams/Oracle-to-Source",
  "read_method": "oracle-cdc-logminer",
  "object": "SAMPLE.TBL",
  "uuid": "d7989206-380f-0e81-8056-240501101100",
  "read_timestamp": "2019-11-07T07:37:16.808Z",
  "source_timestamp": "2019-11-07T02:15:39",  
  "source_metadata": {
    "log_file": ""
    "scn": 15869116216871,
    "row_id": "AAAPwRAALAAMzMBABD",
    "is_deleted": false,
    "database": "DB1",
    "schema": "ROOT",
    "table": "SAMPLE"
    "change_type": "INSERT",
    "tx_id": 
    "rs_id": "0x0073c9.000a4e4c.01d0",
    "ssn": 67,
  },
  "payload": {
    "THIS_IS_MY_PK": "1231535353",
    "FIELD1": "foo",
    "FIELD2": "TLV",
  }
}

UPDATE (T1)

訊息酬載包含整個新資料列。不含先前的值。

{
  "stream_name": "projects/myProj/locations/myLoc/streams/Oracle-to-Source",
  "read_method": "oracle-cdc-logminer",
  "object": "SAMPLE.TBL",
  "uuid": "e6067366-1efc-0a10-a084-0d8701101101",
  "read_timestamp": "2019-11-07T07:37:18.808Z",
  "source_timestamp": "2019-11-07T02:17:39",  
  "source_metadata": {
    "log_file": 
    "scn": 15869150473224,
    "row_id": "AAAGYPAATAAPIC5AAB",
    "is_deleted": false,
    "database":
    "schema": "ROOT",
    "table": "SAMPLE"
    "change_type": "UPDATE",
    "tx_id":
    "rs_id": "0x006cf4.00056b26.0010",
    "ssn": 0,
  },
  "payload": {
    "THIS_IS_MY_PK": "1231535353",
    "FIELD1": null,
    "FIELD2": "TLV",
  }
}

DELETE (T2)

訊息酬載包含整個新資料列。

{
  "stream_name": "projects/myProj/locations/myLoc/streams/Oracle-to-Source",
  "read_method": "oracle-cdc-logminer",
  "object": "SAMPLE.TBL",
  "uuid": "c504f4bc-0ffc-4a1a-84df-6aba382fa651",
  "read_timestamp": "2019-11-07T07:37:20.808Z",
  "source_timestamp": "2019-11-07T02:19:39",
  "source_metadata": {
    "log_file": 
    "scn": 158691504732555,
    "row_id": "AAAGYPAATAAPIC5AAC",
    "is_deleted": true,
    "database":
    "schema": "ROOT",
    "table": "SAMPLE"
    "change_type": "DELETE",
    "tx_id":
    "rs_id": "0x006cf4.00056b26.0011",
    "ssn": 0,
  },
  "payload": {
    "THIS_IS_MY_PK": "1231535353",
    "FIELD1": null,
    "FIELD2": "TLV",
  }
}

排序和一致性

本節說明 Datastream 如何處理排序和一致性。

排序

Datastream 不保證會依序傳送事件，但每個事件都包含完整的資料列、資料寫入來源的時間戳記，以及排序中繼資料。

屬於初始補充作業的事件，其 read_method 欄位會設為 backfill，或結尾為 -backfill、-fulldump 或 -incremental。補充事件的產生順序不固定，可依任何順序使用。

如果事件屬於進行中的變更資料擷取 (CDC) 複製作業，read_method 欄位會設為 cdc 或來源專屬的 CDC 值，例如 mysql-cdc-binlog、oracle-cdc-logminer、postgres-cdc-wal、sqlserver-cdc 或 salesforce-cdc。如何排序所有來源的持續複製事件，取決於目的地：

目的地 說明
BigQuery

BigQuery 中的事件排序方式取決於您為串流設定的寫入模式：

  • 合併寫入模式：BigQuery 會使用事件中繼資料和內部變更序號 (CSN)，自動依正確順序合併順序不正確的事件。沒有主鍵的資料表會以僅限附加模式寫入。
  • 僅附加寫入模式，包括 Apache Iceberg 資料表：Datastream 會將變更事件附加至目的地資料表，並填入 datastream_metadata STRUCT 欄。如要判斷事件在來源發生的順序，請使用 datastream_metadata.change_sequence_number 欄位或 datastream_metadata.sort_keys 陣列欄位排序事件。
Cloud Storage

在 Cloud Storage 中，相同時間範圍內的事件可能會跨越多個檔案。如要判斷事件在來源發生的順序，請使用一般中繼資料中的 sort_keys 陣列欄位排序事件。

一致性

Datastream 會確保來源資料庫中的資料至少傳送至目的地一次。系統不會遺漏任何事件，但串流中可能會有重複的事件。重複事件的時間範圍應以分鐘為單位，且事件中繼資料中的事件通用不重複 ID (UUID) 可用於偵測重複事件。

不過，請勿只依賴 UUID 來簡化下游事件。如果管道中斷，Datastream 會從上次儲存的檢查點重新讀取來源記錄，確保資料傳送。Datastream 讀取事件時會動態產生 UUID，因此這類重試會為完全相同的來源資料產生新的 UUID。此外，在來源資料庫中執行大量作業時，多個不同的資料列可能會共用相同的變更序號 (CSN)。

如要確實簡化資料，請依來源資料表主鍵劃分事件，並使用變更序號 (CSN) (change_sequence_number 欄位) 將事件分組。這樣一來，系統就能正確收合相同事件的重試作業，同時保留大量交易中合法的不同資料列更新。

如果資料庫記錄檔包含未修訂的交易，且有任何交易復原，資料庫就會在記錄檔中將這項操作反映為「反向」data manipulation language (DML) 作業。例如，復原的 INSERT 作業會有對應的 DELETE 作業。Datastream 會從記錄檔讀取這些作業。

關於串流

每個串流都有中繼資料，用於描述串流和資料來源。這類中繼資料包含串流名稱、來源和目的地連線設定檔等資訊。

如要查看 Stream 物件的完整定義，請參閱 API 參考資料說明文件。

串流狀態

串流可能處於下列其中一種狀態：

  • Not started
  • Starting
  • Running
  • Draining
  • Paused
  • Failed
  • Failed permanently

您可以使用記錄檔查看其他狀態資訊，例如資料表補充或處理的資料列數量。您也可以使用 FetchStreamErrors API 擷取錯誤。

可透過探索 API 取得的物件中繼資料

探索 API 會傳回物件，代表連線設定檔所代表的資料來源或目的地中定義的物件結構。每個物件都有本身的中繼資料，以及所提取每項資料欄位的中繼資料。這項中繼資料可透過 discover API 取得。

後續步驟