事件和数据流

Datastream 中的数据层次结构如下：

  • 数据流，由数据源和目标位置组成。
  • 对象，是数据流的一部分，例如来自特定数据库的表。
  • 事件，由特定对象（例如数据库插入）生成的单个更改。

数据流、对象和事件具有关联的数据和元数据。这些数据和元数据可用于不同目的。

关于事件

每个事件包含 3 种类型的数据：

  • 事件数据：表示源自流来源的对象的数据本身发生的更改。每个事件都包含已更改的整行。
  • 通用元数据：此元数据会显示在 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
  • salesforce-cdc
  • salesforce-backfill
object 字符串 字符串 用于对不同类型的事件进行分组的名称，通常是来源中表或对象的名称。
schema_key 字符串 字符串 事件统一架构的唯一标识符。
uuid 字符串 字符串 Datastream 生成的事件的唯一标识符。
read_timestamp timestamp-millis 字符串 Datastream 读取记录时的时间戳 (UTC)（Epoch 时间戳，以毫秒为单位）。
source_timestamp timestamp-millis 字符串 记录在来源中发生更改时的时间戳 (UTC)（Epoch 时间戳，以毫秒为单位）。
sort_keys {"type": "array", "items": ["string", "long"]} 数组 可用于按事件发生顺序对事件进行排序的值数组。

特定于来源的元数据

此元数据与来自源数据库的 CDC 和回填事件相关联。如需查看此元数据，请从以下下拉菜单中选择一个来源。

来源 字段 Avro 类型 JSON 类型 说明
MySQL log_file 字符串 字符串 使用基于二进制日志的复制时，Datastream 在 CDC 复制中从中拉取事件的日志文件。
MySQL log_position 长整型 长整型 使用基于 binlog 的复制时，MySQL 二进制日志中的日志位置（偏移量）。
MySQL binlog_file 字符串 字符串 使用基于 GTID 的复制时，Datastream 在 CDC 复制中从中拉取事件的二进制日志文件。
MySQL binlog_position 长整型 长整型 使用基于 GTID 的复制时，MySQL 二进制日志中的位置（偏移量）。
MySQL gtid 字符串 字符串 使用基于 GTID 的复制时，用于唯一标识事件事务的全局事务标识符 (GTID)。
MySQL datastream_master_server_uuid 字符串 字符串 使用基于 GTID 的复制时，Datastream 从中读取事件的 MySQL 源服务器实例的 UUID。
MySQL datastream_master_server_id 长整型 长整型 使用基于 GTID 的复制时，Datastream 从中读取事件的 MySQL 源服务器实例的服务器 ID。
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 长整型 长整型 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 长整型 长整型 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 长整型 长整型 与事件关联的数据库。
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 字符串 字符串 事件的提交时间戳。
Spanner snapshot 布尔值 布尔值 事件是否为回填快照事件。
Spanner project_id 字符串 字符串 Spanner 项目标识符。
Spanner instance_id 字符串 字符串 Spanner 实例标识符。
Spanner database_id 字符串 字符串 Spanner 数据库标识符。
Spanner change_stream_name 字符串 字符串 Spanner 变更数据流。
Spanner table 字符串 字符串 Spanner 表。
Spanner server_transaction_id 字符串 字符串 与事件关联的交易标识符。
Spanner record_sequence 字符串 字符串 与事件关联的记录序列。
Spanner mod_index 字符串 字符串 与事件关联的修改编号。
Spanner transaction_tag 字符串 字符串 关联事务的标记。
Spanner system_transaction 字符串 字符串 相应事务是否为系统事务。
Spanner number_of_records_in_transaction 长整型 长整型 关联交易中的记录数。
Spanner value_capture_type 枚举 枚举 变更数据流的值捕获类型。
Spanner mod_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 字段，用作集合中每个文档的主键。

事件流示例

此流说明了由 3 个连续操作（INSERT、UPDATE 和 DELETE）在源数据库的 SAMPLE 表中的一行上生成的事件。

时间 THIS_IS_MY_PK (int) FIELD1 (nchar nullable) FIELD2 (nchar non-null)>
0 1231535353 foo TLV
1 1231535353 NULL 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 不保证顺序，但每个事件都包含完整的数据行以及数据写入源的时间戳。在 BigQuery 中，乱序事件会自动按正确顺序合并。BigQuery 会使用事件元数据和内部更改序列号 (CSN)，按正确的顺序将事件应用于表。在 Cloud Storage 中，同一时间的事件可能会跨越多个文件。

按照设计，在为了初始化数据流时创建的数据的初始回填而对事件进行回填时，会无序生成事件。

可以根据每个来源推断顺序。

来源 说明
MySQL

属于初始回填的事件的 read_method 字段以 mysql-backfill 开头（对于基于 binlog 的复制）或设置为 backfill（对于基于 GTID 的复制）。回填中接收事件的顺序没有任何影响，因为它们可以按任何顺序使用。

属于正在进行的复制的事件会将 read_method 字段设置为 mysql-cdc-binlog（对于基于 binlog 的复制）或 cdc（对于基于 GTID 的复制）。

对于基于 binlog 的复制，可以通过 log_file 和 log_position 字段的组合来推断顺序。对于基于 GTID 的复制，可以通过 binlog_file 和 binlog_position 字段（或 gtid 字段）的组合来推断顺序。这种组合提供了一个唯一的、递增的数字，用于标识数据库中的操作顺序。
Oracle

属于初始回填的事件的 read_method 字段以 oracle-backfill 开头。在回填中接收事件的顺序不受影响，因为可以按任意顺序使用它们。

属于正在进行的复制的事件将 read_method 字段设置为 oracle-cdc-logminer。

该顺序可由 rs_id（记录集 ID）字段和 ssn（SQL 序列号）字段的组合推断得出。此组合提供一个唯一的递增编号，用于标识数据库中的操作顺序。
PostgreSQL

属于初始回填的事件的 read_method 字段以 postgresql-backfill 开头。在回填中接收事件的顺序不受影响，因为可以按任意顺序使用它们。

属于正在进行的复制的事件将 read_method 字段设置为 postgres-cdc-wal。

顺序可以通过 source_timestamp 字段和 lsn（日志序列号）字段的组合来推断。此组合提供一个唯一的递增编号，用于标识数据库中的操作顺序。
SQL Server

属于初始回填的事件的 read_method 字段以 sqlserver-backfill 开头。在回填中接收事件的顺序不受影响，因为可以按任意顺序使用它们。

属于正在进行的复制的事件会将 read_method 字段设置为 sqlserver-cdc。

顺序可以通过 source_timestamp 字段和 lsn（日志序列号）字段的组合来推断。此组合提供一个唯一的递增编号，用于标识数据库中的操作顺序。

复制到 BigQuery 时，数据库创建时间也用于确保顺序正确。这在源数据库实例可能发生变化的情况下（例如在迁移或故障切换到副本期间）非常重要。
Salesforce

您可以使用记录的 source_timestamp 作为排序键来确定顺序。Salesforce 中的时间戳具有一秒的分辨率，但同一记录不能在同一秒发生两个更改事件。
MongoDB

顺序可以通过使用操作日志中的 ts 字段或记录的变更数据流中的 clusterTime 字段来确定。这些字段对于每条记录都是唯一的。
Spanner

顺序可以通过 commit_timestamp、record_sequence 和 mod_index 字段的组合来确定。这种组合提供了一个唯一的、递增的值，用于标识数据库中的操作顺序。

一致性

Datastream 可确保源数据库中的数据至少一次传送到目标位置。不会遗漏任何事件，但数据流中可能会出现重复事件。重复事件的时段应大约几分钟，事件元数据中事件的通用唯一标识符 (UUID) 可用于检测重复项。

不过，请勿仅依靠 UUID 来对下游事件进行去重。如果流水线发生中断，Datastream 会从上次保存的检查点重新读取源日志，以确保数据成功传送。由于 Datastream 在读取事件时会动态生成 UUID，因此这种重试会导致完全相同的源数据生成新的 UUID。此外，在源数据库中执行批量操作期间，多个不同的行可能会共享相同的变更序列号 (CSN)。

如需可靠地对数据进行去重，请按源表主键对事件进行分区，并使用变更序列号 (CSN)（change_sequence_number 字段）对事件进行分组。这样可以确保正确合并同一事件的重试，同时保留批量事务中合法的不同行更新。

当数据库日志文件包含未提交的事务时，如果回滚了任何事务，则数据库会在日志文件中将其反映为“反向”数据操纵语言 (DML) 操作。例如，回滚 INSERT 操作具有相应的 DELETE 操作。Datastream 从日志文件中读取这些操作。

关于数据流

每个数据流都具有描述该数据流及其数据来源的元数据。此类元数据包含诸如数据流名称、来源和目标连接配置文件之类的信息。

如需查看数据流对象的完整定义，请查看 API 参考文档。

数据流状态

数据流可能处于以下任一状态：

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

您可以使用日志查找其他状态信息，例如回填的表或处理的行数。您还可以使用 FetchStreamErrors API 检索错误。

可通过发现 API 使用的对象元数据

发现 API 会返回一些对象，这些对象表示在连接配置文件所表示的数据源或目标中定义的对象的结构。每个对象都有关于对象本身的元数据，以及它提取的每个数据字段的元数据。您可以使用发现 API 获取此元数据。

后续步骤