更改流分区、记录和查询

本页面详细介绍了更改流的以下属性：

  • 其基于拆分的分区模型
  • 变更流记录的格式和内容
  • 用于查询这些记录的低级语法
  • 查询工作流示例

本页中的信息与使用 Spanner API 直接查询更改流最相关的。 改为使用 Dataflow 读取更改流数据的应用不需要直接使用本文所述的数据模型。

如需查看更详尽的数据流更改入门指南，请参阅更改数据流概览。

更改流分区

当更改流所监控的表发生更改时，Cloud Spanner 会在数据库中写入相应的更改流记录，并在与数据更改相同的事务中同步该记录。这可确保在事务成功后，Spanner 也已成功捕获并保留相应更改。在内部，Spanner 会将变更流记录和数据更改存储在一起，以便由同一服务器处理，从而最大限度地减少写入开销。

作为特定 DML 的一部分，Spanner 会将写入操作附加到同一事务中的相应更改流数据拆分中。由于存在这种共存，变更流不会跨服务资源增加额外的协调，从而最大限度地降低事务提交开销。

映像

Spanner 根据数据库负载和大小动态拆分和合并数据，并将分片拆分到各个服务资源中，从而实现扩容。

为了支持变更流的写入和读取扩缩，Spanner 会拆分内部变更流的存储空间以及数据库数据，以自动避免热点。为支持在数据库写入规模时近乎实时地读取更改流记录，Spanner API 旨在使用更改流分区并发查询更改流。更改流分区映射可更改包含变更流记录的数据流数据拆分。变更流的分区会随时间动态变化，并且与 Spanner 动态拆分和合并数据库数据的方式相关。

变更流分区包含特定时间范围内不可变键范围的记录。任何变更流分区都可以拆分为一个或多个变更流分区，也可以与其他变更流分区合并。当发生这些拆分或合并事件时，系统会创建子分区，以捕获下一个范围内的不可变键范围的更改。除了数据更改记录之外，变更流查询还会返回子分区记录以通知需要查询的新变更流分区，以及检测信号记录以指示最近没有发生任何写入操作时向前推进。

查询特定更改流分区时，系统会按照提交时间戳的顺序返回更改记录。每个变更记录只返回一次。在各个更改流分区中，不能保证更改记录的顺序。特定主键的变更记录仅针对特定时间范围内的一个分区返回。

由于父子分区的沿袭，为了按提交时间戳顺序处理特定键的更改，应仅在处理完所有父分区的记录后才处理从子分区返回的记录。

更改数据流查询语法

使用 ExecuteStreamingSql API 查询更改流。系统会自动为变更流创建一个特殊的表值函数 (TVF)。提供对变更流记录的访问权限。TVF 的命名惯例为 READ_change_stream_name。

假设数据库中已存在更改流 SingersNameStream，则查询语法如下所示：

SELECT ChangeRecord
    FROM READ_SingersNameStream (
        start_timestamp,
        end_timestamp,
        partition_token,
        heartbeat_milliseconds
    )

该函数接受以下参数：

参数名称 类型 是否必需？ 说明
start_timestamp TIMESTAMP 必填 指定返回 commit_timestamp 大于或等于 start_timestamp 的记录。该值必须在变更流保留期限内，且应小于或等于当前时间，并大于或等于变更流创建时间戳。
end_timestamp TIMESTAMP 可选（默认值：NULL） 指定返回 commit_timestamp 小于或等于 end_timestamp 的记录。该值必须在变更流保留期限内，并且大于或等于 start_timestamp。查询将在返回 end_timestamp 之前的所有 ChangeRecord 或返回一组子分区记录后完成。如果未指定 NULL，则系统将执行查询，直到完成当前分区并返回所有设置了 child_partition_record 字段的 ChangeRecord。为 end_timestamp 指定 NULL 表示始终读取最新的更改。
partition_token STRING 可选（默认值：NULL） 根据子分区记录的内容指定要查询的变更流分区。如果未指定或未指定 NULL，则表示读取器首次查询变更流，并且尚未获取任何特定分区令牌进行查询。
heartbeat_milliseconds INT64 必填 确定在此分区中未提交任何事务时返回检测信号 ChangeRecord 的频率。该值必须介于 1000（一秒）和 300000（五分钟）之间。

我们建议采用一种便捷方法构建 TVF 查询的文本并将其绑定到文本，如以下示例所示。

Java

private static final String SINGERS_NAME_STREAM_QUERY_TEMPLATE =
        "SELECT ChangeRecord FROM READ_SingersNameStream"
            + "("
            + "   start_timestamp => @startTimestamp,"
            + "   end_timestamp => @endTimestamp,"
            + "   partition_token => @partitionToken,"
            + "   heartbeat_milliseconds => @heartbeatMillis"
            + ")";

// Helper method to conveniently create change stream query texts and bind parameters.
public static Statement getChangeStreamQuery(
      String partitionToken,
      Timestamp startTimestamp,
      Timestamp endTimestamp,
      long heartbeatMillis) {
    return Statement.newBuilder(SINGERS_NAME_STREAM_QUERY_TEMPLATE)
                    .bind("startTimestamp")
                    .to(startTimestamp)
                    .bind("endTimestamp")
                    .to(endTimestamp)
                    .bind("partitionToken")
                    .to(partitionToken)
                    .bind("heartbeatMillis")
                    .to(heartbeatMillis)
                    .build();
}

更改视频流记录格式

变更流 TVF 会返回类型为 ARRAY<STRUCT<...>> 的单个 ChangeRecord 列。在每行中，该数组始终只包含一个元素。

数组元素具有以下类型：

STRUCT <
  data_change_record ARRAY<STRUCT<...>>,
  heartbeat_record ARRAY<STRUCT<...>>,
  child_partitions_record ARRAY<STRUCT<...>>
>

此结构中有三个字段：data_change_record、heartbeat_record 和 child_partitions_record，每个字段的类型均为 ARRAY<STRUCT<...>>。在变更流 TVF 返回的任何行中，只有一个字段包含三个值；另外两个字段为空或 NULL。这些数组字段最多只能包含一个元素。

以下部分将分别介绍这三种记录类型。

数据更改记录

数据更改记录包含对同一事务的一个更改流分区中在相同提交时间戳处提交的修改类型（插入、更新或删除）的一组更改。可以针对多个更改流分区中的同一事务返回多个数据更改记录。

所有数据更改记录都包含 commit_timestamp、server_transaction_id 和 record_sequence 字段，它们共同决定了流记录的更改流中的顺序。这三个字段足以推导出更改的顺序并提供外部一致性。

请注意，如果多个事务触摸非重叠数据，它们可以具有相同的提交时间戳。server_transaction_id 字段用于区分哪些更改（可能跨越变更流分区）在同一事务中发出。将其与 record_sequence 和 number_of_records_in_transaction 字段配对可让您缓冲来自特定事务的所有记录并对其进行排序。

数据更改记录的字段包括以下内容：

字段 类型 说明
commit_timestamp TIMESTAMP 表示提交更改的时间戳。
record_sequence STRING 事务中记录的序列号。在事务中，序列号保证是唯一的单调递增（但不一定是连续的）。按“记录序列”对同一“server_transaction_id”的记录进行排序，以在事务内重建更改的顺序。
server_transaction_id STRING 表示提交更改的事务的全局唯一字符串。此值应仅在处理更改流记录的情况下使用，与 Spanner API 中的事务 ID（例如 `TransactionSelector.id`）无关。当与同一上下文中的其他值（即更改流 `data_change_records` 或 Spanner API）相比，这两项都能唯一标识事务。
is_last_record_in_transaction_in_partition BOOL 指明这是否为当前分区中事务的最后一条记录。
table_name STRING 受更改影响的表的名称。
value_capture_type STRING

描述捕获此变更时在变更流配置中指定的值捕获类型。

目前始终为 "OLD_AND_NEW_VALUES"。
column_types ARRAY<STRUCT<
name STRING,
 type JSON,
 is_primary_key BOOL,
 ordinal_position INT64
>>		 列的名称、列类型、是否为主键，以及架构中定义的列的位置 (`ordinal_position`)。表的第一列在架构中的顺序编号位置将为 `1`。列类型可以嵌套在数组列中。格式与 Spanner API 参考文档中所述的类型结构相匹配。
mods ARRAY<STRUCT<
keys JSON,
 new_values JSON,
 old_values JSON
>>		 说明所做的更改，包括主键值以及已更改的列的旧值和新值（如果配置了“value_capture_type”）。new_values 和 old_values 字段仅包含非键列。
mod_type STRING 描述更改的类型。INSERT、UPDATE 或 DELETE 中的一个。
number_of_records_in_transaction INT64 在所有变更流分区中，属于此事务的数据变更记录数量。
number_of_partitions_in_transaction INT64 将返回此事务的数据更改记录的分区数。

以下是一对示例数据更改记录。用于描述两个帐号之间发生转移作业的单个事务。请注意，这两个帐号位于单独的变更流分区中。

data_change_record: {
  "commit_timestamp": "2022-09-27T12:30:00.123456Z",
  // record_sequence is unique and monotonically increasing within a
  // transaction, across all partitions.
  "record_sequence": "00000000",
  "server_transaction_id": "6329047911",
  "is_last_record_in_transaction_in_partition": true,

  "table_name": "AccountBalance",
  "column_types": [
    {
      "name": "AccountId",
      "type": {"code": "STRING"},
      "is_primary_key": true,
      "ordinal_position": 1
    },
    {
      "name": "LastUpdate",
      "type": {"code": "TIMESTAMP"},
      "is_primary_key": false,
      "ordinal_position": 2
    },
    {
       "name": "Balance",
       "type": {"code": "INT"},
       "is_primary_key": false,
       "ordinal_position": 3
    }
  ],
  "mods": [
    {
      "keys": {"AccountId": "Id1"},
      "new_values": {
        "LastUpdate": "2022-09-27T12:30:00.123456Z",
        "Balance": 1000
      },
         "old_values": {
        "LastUpdate": "2022-09-26T11:28:00.189413Z",
        "Balance": 1500
      },
    }
  ],
  "mod_type": "UPDATE", // options are INSERT, UPDATE, DELETE
  "value_capture_type": "OLD_AND_NEW_VALUES",
  "number_of_records_in_transaction": 2,
  "number_of_partitions_in_transaction": 2,
}

data_change_record: {
  "commit_timestamp": "2022-09-27T12:30:00.123456Z",
  "record_sequence": "00000001",
  "server_transaction_id": "6329047911",
  "is_last_record_in_transaction_in_partition": true,

  "table_name": "AccountBalance",
  "column_types": [
    {
      "name": "AccountId",
      "type": {"code": "STRING"},
      "is_primary_key": true,
      "ordinal_position": 1
    },
    {
      "name": "LastUpdate",
      "type": {"code": "TIMESTAMP"},
      "is_primary_key": false,
      "ordinal_position": 2
    },
    {
      "name": "Balance",
      "type": {"code": "INT"},
      "is_primary_key": false,
      "ordinal_position": 3
    }
  ],
  "mods": [
    {
      "keys": {"AccountId": "Id2"},
      "new_values": {
        "LastUpdate": "2022-09-27T12:30:00.123456Z",
        "Balance": 2000
      },
      "old_values": {
        "LastUpdate": "2022-01-20T11:25:00.199915Z",
        "Balance": 1500
      },
    },
    ...
  ],
  "mod_type": "UPDATE", // options are INSERT, UPDATE, DELETE
  "value_capture_type": "OLD_AND_NEW_VALUES",
  "number_of_records_in_transaction": 2,
  "number_of_partitions_in_transaction": 2,
}

检测信号记录

返回检测信号记录时，表示 commit_timestamp 小于或等于检测信号记录 timestamp 的所有更改均已返回，且此分区中未来数据记录的提交时间戳必须高于检测信号记录返回的时间戳。当没有数据写入分区时，会返回检测信号记录。当有数据写入分区时，可以使用 data_change_record.commit_timestamp 代替 heartbeat_record.timestamp，以告知读取器正在读取分区。

您可以使用分区上返回的检测信号记录来同步所有分区中的读取器。一旦所有读取者都收到大于或等于时间戳 A 的检测信号，或者收到大于或等于时间戳 A 的数据或子分区记录，读取器便会知道已收到在时间戳 A 或之前提交的所有记录，并且可以开始处理已缓冲的记录 - 例如，按时间戳对跨分区记录进行排序并按 server_transaction_id 对它们进行分组。

检测信号记录仅包含一个字段：

字段 类型 说明
timestamp TIMESTAMP 检测信号记录时间戳。

检测信号记录示例，指示返回了所有不超过此时间戳的时间戳的记录：

heartbeat_record: {
  "timestamp": "2022-09-27T12:35:00.312486Z"
}

子分区记录

子分区记录会返回有关子分区的信息：其分区令牌、其父分区的令牌，以及表示子分区包含更改记录的最早时间戳的 start_timestamp。其提交时间戳紧挨着 child_partitions_record.start_timestamp 的记录会返回当前分区中。返回此分区的所有子分区记录后，此查询将返回成功状态，表示此分区已返回所有记录。

子分区记录的字段包括以下内容：

字段 类型 说明
start_timestamp TIMESTAMP 从此子分区记录中的子分区返回的数据更改记录具有大于或等于 start_timestamp 的提交时间戳。查询子分区时，查询应指定子分区令牌以及大于或等于 child_partitions_token.start_timestamp 的 start_timestamp。一个分区返回的所有子分区记录将具有相同的 start_timestamp，且时间戳始终在查询指定的 start_timestamp 和 end_timestamp 之间。
record_sequence STRING 单调递增的序列号，可用于在特定分区返回多个具有相同 start_timestamp 的子分区记录时，定义子分区记录的顺序。分区令牌 start_timestamp 和 record_sequence 可唯一标识子分区记录。
child_partitions ARRAY<STRUCT<
token STRING,
parent_partition_tokens
ARRAY<STRING>
>>		 返回一组子分区及其相关信息。这包括在查询中标识子分区的分区令牌字符串，以及其父分区的令牌。

子分区记录示例：

child_partitions_record: {
  "start_timestamp": "2022-09-27T12:40:00.562986Z",
  "record_sequence": "00000001",
  "child_partitions": [
    {
      "token": "child_token_1",
      // To make sure changes for a key is processed in timestamp
      // order, wait until the records returned from all parents
      // have been processed.
      "parent_partition_tokens": ["parent_token_1", "parent_token_2"],
    }
  ],
}

更改流查询工作流

使用 ExecuteStreamingSql API 运行变更流查询，并提供一次性只读事务和强大的时间戳边界。变更流 TVF 允许用户为感兴趣的时间范围指定 start_timestamp 和 end_timestamp。保留期限内的所有更改记录都可以使用强只读时间戳边界访问。

所有其他 TransactionOptions 均不适用于更改流查询。此外，如果 TransactionOptions.read_only.return_read_timestamp 设置为 true，系统将在描述事务的 Transaction 消息中返回一个特殊值 kint64max - 1，而不是有效的读取时间戳。此特殊值应舍弃，不应用于任何后续查询。

每个更改流查询都可以返回任意数量的行，每行都包含数据更改记录、检测信号记录或子分区记录。您无需为请求设置时限。

示例：

流式查询工作流的起始点是指定 partition_token 到 NULL，从而发出第一个更改流查询。该查询需要为变更流、感兴趣的开始和结束时间戳以及检测信号间隔指定 TVF 函数。当 end_timestamp 为 NULL 时，查询将持续返回数据更改，直到子分区生成。

SELECT ChangeRecord FROM READ_SingersNameStream(
  start_timestamp => "2022-05-01 09:00:00-00",
  end_timestamp => NULL,
  partition_token => NULL,
  heartbeat_milliseconds => 10000
);

处理此查询中的数据记录，直到返回子分区记录。在以下示例中，系统会返回两条子分区记录和三个分区令牌，然后查询会终止。特定查询的子分区记录将始终共用相同的 start_timestamp。

child_partitions_record: {
  "record_type": "child_partitions",
  "start_timestamp": "2022-05-01 09:00:01-00",
  "record_sequence": 1000012389,
  "child_partitions": [
    {
      "token": "child_token_1",
      // Note parent tokens are null for child partitions returned
      // from the initial change stream queries.
      "parent_partition_tokens": [NULL],
    }
    {
      "token": "child_token_2",
      "parent_partition_tokens": [NULL],
    }
  ],
}

child partitions record: {
  "record_type": "child_partitions",
  "start_timestamp": "2022-05-01 09:00:01-00",
  "record_sequence": 1000012390,
  "child_partitions": [
    {
      "token": "child_token_3",
      "parent_partition_tokens": [NULL],
    }
  ],
}

如需在 2022-05-01 09:00:01-00 之后处理将来的更改，请创建三个新查询并并行运行它们。这三个查询将同时返回其父级所覆盖的相同键范围的未来数据更改。请始终将同一子分区记录中的 start_timestamp 设置为 start_timestamp，并使用相同的 end_timestamp 和检测信号间隔来处理所有查询的记录。

SELECT ChangeRecord FROM READ_SingersNameStream(
  start_timestamp => "2022-05-01 09:00:01-00",
  end_timestamp => NULL,
  partition_token => "child_token_1",
  heartbeat_milliseconds => 10000);

SELECT ChangeRecord FROM READ_SingersNameStream(
  start_timestamp => "2022-05-01 09:00:01-00",
  end_timestamp => NULL,
  partition_token => "child_token_2",
  heartbeat_milliseconds => 10000);

SELECT ChangeRecord FROM READ_SingersNameStream(
  start_timestamp => "2022-05-01 09:00:01-00",
  end_timestamp => NULL,
  partition_token => "child_token_3",
  heartbeat_milliseconds => 10000);

一段时间后，对 child_token_2 的查询在返回另一个子分区记录后完成，此记录表明，从 2022-05-01 09:30:15-00 开始，新分区将涵盖 child_token_2 和 child_token_3 的未来更改。查询将在 child_token_3 上返回完全相同的记录，因为两者都是新 child_token_4 的父分区。如需保证对特定键的数据记录进行严格排序处理，必须仅在所有父项都完成后查询 child_token_4（在本例中为 child_token_2 和 child_token_3）。只能为每个子分区令牌创建一个查询，查询工作流设计应指定一名父级来等待并在 child_token_4 上安排查询。

child partitions record: {
  "record_type": "child_partitions",
  "start_timestamp": "2022-05-01 09:30:15-00",
  "record_sequence": 1000012389,
  "child_partitions": [
    {
      "token": "child_token_4",
      "parent_partition_tokens": [child_token_2, child_token_3],
    }
  ],
}

SELECT ChangeRecord FROM READ_SingersNameStream(
  start_timestamp => "2022-05-01 09:30:15-00",
  end_timestamp => NULL,
  partition_token => "child_token_4",
  heartbeat_milliseconds => 10000
);

您可以在 GitHub 上查找 Apache Beam SpannerIO Dataflow 连接器中处理和解析更改流记录的示例。