事件和数据流

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
  • sqlserver-cdc
  • sqlserver-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 timestamp-micros 字符串 事件的提交时间戳。
Spanner snapshot 布尔值 布尔值 事件是否为回填快照事件。
Spanner project_id 字符串 字符串 Spanner 项目标识符。
Spanner instance_id 字符串 字符串 Spanner 实例标识符。
Spanner database_id 字符串 字符串 Spanner 数据库标识符。
Spanner schema 字符串 字符串 与事件中的表关联的架构。如果表不在命名架构中，则为空。
Spanner table 字符串 字符串 Spanner 表。
Spanner change_stream_name 字符串 字符串 Spanner 变更数据流。
Spanner server_transaction_id 字符串 字符串 与事件关联的交易标识符。
Spanner record_sequence long 或 string long 或 string 与事件关联的记录序列。对于使用 IMMUTABLE_KEY_RANGE 分区模式创建的变更数据流，该值为 long 类型的数字序列号。对于使用 MUTABLE_KEY_RANGE 分区模式创建的变更数据流，该值是数据更改记录中的 record_sequence 字符串，例如 1cac8f3f38eeaf53-00000000。
Spanner mod_number 长整型 长整型 与事件关联的修改编号。
Spanner transaction_tag 字符串 字符串 关联事务的标记。
Spanner number_of_records_in_transaction 长整型 长整型 关联交易中的记录数。
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 字段，用作集合中每个文档的主键。

事件流示例

此流说明了由 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 不保证按顺序传送事件，但每个事件都包含完整的数据行、数据写入来源的时间戳以及排序元数据。

属于初始回填的事件会将 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 可确保源数据库中的数据至少一次传送到目标位置。不会遗漏任何事件，但数据流中可能会出现重复事件。重复事件的时段应大约几分钟，事件元数据中事件的通用唯一标识符 (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 获取此元数据。

后续步骤