了解 CDC 记录

Aurora DSQL CDC 将每项更改作为一条 JSON 记录进行传输。该记录使用信封结构，包含操作类型、更改前后的行映像以及源元数据。

记录映射到 Amazon Kinesis 的方式

Aurora DSQL 将每条 CDC 记录作为一条 Kinesis 记录写入。Kinesis 记录的 Data 字段包含 JSON 有效载荷。Aurora DSQL 使用随机化的 Kinesis 分区键将 CDC 记录均匀分布到各个分片中。要读取所有更改，请使用 Kinesis 数据流上的所有分片。如果一条记录超过 Kinesis 记录大小限制，Aurora DSQL 会将该记录拆分为多条 Kinesis 记录。有关更多信息，请参阅 处理超大记录。

有效载荷中的主键

对于带有主键的表，主键列值将显示在有效载荷中：

以下面这个带复合主键的表为例：

CREATE TABLE order_items (
    order_id INT,
    item_id INT,
    quantity INT,
    price NUMERIC,
    PRIMARY KEY (order_id, item_id)
);

在此表中执行删除操作会生成有效载荷，其中 "before": {"order_id": 1001, "item_id": 42}。

记录有效载荷

有效载荷使用以下 JSON 信封格式。

INSERT 示例

以下示例显示插入操作的 CDC 记录：

{
    "type": "full",
    "op": "c",
    "before": null,
    "after": {"order_id": 1001, "item_id": 42, "quantity": 5, "price": "29.99"},
    "source": {
        "version": "1.0",
        "ts_ms": 1705318200000,
        "ts_ns": 1705318200000000000,
        "txId": "ffthunp5stx6ffs2vyfqoatmfu",
        "schema": "public",
        "table": "order_items",
        "db": "postgres",
        "cluster": "kmabugltfmjdaj2siqr2qbxgju"
    },
    "ts_ms": 1705318200125,
    "ts_ns": 1705318200125483291
}

UPDATE 示例

以下示例显示在 Aurora DSQL 开始发出 op: "u" 后，由 UPDATE 语句生成的 CDC 记录的外观：

{
    "type": "full",
    "op": "u",
    "before": null,
    "after": {"order_id": 1001, "item_id": 42, "quantity": 10, "price": "29.99"},
    "source": {
        "version": "1.0",
        "ts_ms": 1705318300000,
        "ts_ns": 1705318300000000000,
        "txId": "qvtiesgmd55cvlfukm3dfuotji",
        "schema": "public",
        "table": "order_items",
        "db": "postgres",
        "cluster": "kmabugltfmjdaj2siqr2qbxgju"
    },
    "ts_ms": 1705318300125,
    "ts_ns": 1705318300125483291
}

DELETE 示例

在带主键的表中执行删除操作时，before 字段将包含已删除的行的主键值：

{
    "type": "full",
    "op": "d",
    "before": {"order_id": 1001, "item_id": 42},
    "after": null,
    "source": {
        "version": "1.0",
        "ts_ms": 1705318400000,
        "ts_ns": 1705318400000000000,
        "txId": "xyzabc123def456ghi789jklmno",
        "schema": "public",
        "table": "order_items",
        "db": "postgres",
        "cluster": "kmabugltfmjdaj2siqr2qbxgju"
    },
    "ts_ms": 1705318400125,
    "ts_ns": 1705318400125483291
}

有效载荷字段

格式详细信息

以下详细信息描述了 Aurora DSQL CDC 设置记录格式的方式。设计应用程序以处理这些行为。

处理超大记录

当 CDC 记录的序列化 JSON 超过 9 MiB 时，Aurora DSQL 会拆分 before 和/或 after 映像，并传输多条 Kinesis 记录。每条记录均包含一个顶层的 type 字段，该字段可指示记录的结构：full 表示完整记录、chunked 表示引用片段的主记录，fragment 表示分块映像的单个片段。分块主记录上的 op、source、ts_ms 和 ts_ns 字段的行为与其在完整记录上的行为相同。单条 Kinesis 记录中包含的记录的 type 已设置为 full，无需任何额外处理。

chunk_id 在重试期间保持稳定。如果 Aurora DSQL 重新传输某个片段，该片段将携带与原始传输相同的 chunk_id，这样一来，您的应用程序便能在同一标识符下继续缓冲，而无需处理来自先前尝试的部分集合。

主记录

分块的主记录将拆分映像的顶层 before 或 after 字段替换为一个 chunked 对象，该对象描述如何重组映像。chunked 下方的每个条目均包含 chunk_id（将片段链接到此记录的标识符）、total_fragments（组成该映像的片段数量）和 crc32c（重组后的映像文本的 CRC32C 校验和，以十进制字符串形式表示）。如果一个映像是内联的，而另一个映像是分块的，则内联映像仍以值或 null 的形式显示在顶层。

{
    "type": "chunked",
    "op": "c",
    "before": null,
    "after": null,
    "source": {
        "version": "1.0",
        "ts_ms": 1705318200000,
        "ts_ns": 1705318200000000000,
        "txId": "ffthunp5stx6ffs2vyfqoatmfu",
        "schema": "public",
        "table": "order_items",
        "db": "postgres",
        "cluster": "cluster-id"
    },
    "chunked": {
        "after": {
            "chunk_id": "chunk-id",
            "total_fragments": 3,
            "crc32c": "2073618257"
        }
    },
    "ts_ms": 1705318200125,
    "ts_ns": 1705318200125483291
}

片段记录

每个片段均为其自己的 Kinesis 记录，其 type 设置为 fragment，并包含三个字段：chunk_id 匹配主记录上相应 chunked.before.chunk_id 或 chunked.after.chunk_id 中的值，index 是片段在映像中的位置（从零开始计数），data 是按 UTF-8 字符边界拆分的映像 JSON 文本片段（每个片段的 data 值本身是有效的 UTF-8 字符串）。由于 Aurora DSQL CDC 使用 UNORDERED 模式和随机分区键，因此片段和主记录可按任意顺序到达不同的分片。要读取所有片段，请使用 Kinesis 数据流上的所有分片。有关传输排序的更多信息，请参阅排序。

{
    "type": "fragment",
    "chunk_id": "chunk-id",
    "index": 0,
    "data": "partial-JSON-text"
}

要重组超大映像，请按 chunk_id 缓冲每条 type 为 fragment 的记录。在收到一条 type 为 chunked 的主记录时，请等待，直到对于 chunked.before 或 chunked.after 下引用的每个 chunk_id 获得 total_fragments 个片段，按 index 升序对片段进行排序，然后将 data 字符串连接起来。连接后的结果是原始 before 或 after 对象（JSON 文本格式），解析该对象即可访问列值。要验证传输完整性，请在连接后的字符串上计算 CRC32C，然后将结果与 chunked.before.crc32c 或 chunked.after.crc32c 进行比较。

数据类型序列化

下表描述 Aurora DSQL 如何序列化 CDC 记录中的每种 PostgreSQL 数据类型。

整数类型

在 JavaScript 环境中，超出 ±2^53 的 bigint 值可能会丢失精度。在这种情况下，请使用 BigInt 或任意精度库。

浮点类型

布尔值

字符类型

二元

日期和时间类型

其他类型

NULL 值

对于任何数据类型，NULL 列值都表示为 JSON null。

CDC 记录中的架构演变

在修改表的架构时（例如，通过添加、删除或重命名列），CDC 记录会反映自提交 DDL 更改的事务开始的更改。在 DDL 更改之前提交的事务产生的记录使用以前的架构。例如：

通过检查每条记录的 after 和 before 字段中存在的列名，跟踪下游使用者的架构更改。每条记录中的 source.version 字段均标识 CDC 信封格式。