7.3. Debezium PostgreSQL 连接器数据更改事件的描述

Debezium PostgreSQL 连接器为每个行级 INSERT、UPDATE 和 DELETE 操作生成数据更改事件。每个事件包含一个键和值。键的结构和值取决于已更改的表。

Debezium 和 Kafka Connect 围绕 事件消息的持续流 设计。但是，这些事件的结构可能会随时间推移而改变，而用户很难处理这些事件。要解决这个问题，每个事件都包含其内容的 schema，或者如果您正在使用 schema registry，用户可以使用该模式 ID 从 registry 获取 schema。这使得每个事件都自包含。

以下框架 JSON 显示更改事件的基本四部分。但是，如何配置您选择在应用程序中使用的 Kafka Connect converter，决定更改事件中的这四个部分的表示。只有在将转换器配置为生成它时，schema 字段才会处于更改事件中。同样，只有在您配置转换器来生成它时，事件密钥和事件有效负载才会处于更改事件中。如果您使用 JSON 转换程序，并将其配置为生成所有四个基本更改事件部分，更改事件具有此结构：

{
 "schema": { 1
   ...
  },
 "payload": { 2
   ...
 },
 "schema": { 3
   ...
 },
 "payload": { 4
   ...
 },
}

表 7.7. 更改事件基本内容概述
项	字段名称	描述
1	`schema`	第一个 `schema` 字段是事件键的一部分。它指定一个 Kafka Connect 模式，用于描述事件键的 `payload` 部分的内容。换句话说，第一个 `schema` 字段描述了主密钥的结构，如果表没有主键，则描述主键的结构。可以通过设置 `message.key.columns` 连接器配置属性来覆盖表的主键。在这种情况下，第一个 schema 字段描述了该属性标识的键的结构。
2	`payload`	第一个 `payload` 字段是 event 键的一部分。它具有前面的 `schema` 字段描述的结构，它包含已更改的行的密钥。
3	`schema`	第二个 `schema` 字段是事件值的一部分。它指定 Kafka Connect 模式，用于描述事件值 `有效负载部分的内容`。换句话说，第二个 `模式` 描述了已更改的行的结构。通常，此模式包含嵌套模式。
4	`payload`	第二个 `payload` 字段是事件值的一部分。它具有上一个 `schema` 字段描述的结构，它包含已更改的行的实际数据。

默认的行为是，连接器流将事件记录更改为名称与事件原始表相同的主题。

注意

从 Kafka 0.10 开始，Kafka 可以选择使用创建消息 的时间戳 （由 producer 记录）或 Kafka 写入日志的时间戳记录事件键和值。

警告

PostgreSQL 连接器确保所有 Kafka Connect 模式名称都遵循 Avro 模式名称格式。这意味着逻辑服务器名称必须以拉丁字母或下划线开头，即 a-z、A-Z 或 _。逻辑服务器名称和 schema 和表名称中的每个字符都必须是一个拉丁字母、数字或下划线，即 a-z、A-Z、0-9 或 \_。如果存在无效字符，它将使用下划线字符替换。

如果逻辑服务器名称、模式名称或表名称包含无效字符，且唯一与另一个名称区分名称的字符无效，这可能会导致意外冲突冲突，从而被下划线替换。

详情包括在以下主题中：

7.3.1. 关于 Debezium PostgreSQL 中的键更改事件

对于给定表，更改事件的键具有结构，该结构在表的主键中包含创建事件时每个列的字段。或者，如果表将 REPLICA IDENTITY 设置为 FULL 或 USING INDEX，则每个唯一键约束都有一个字段。

考虑在 公共 数据库架构中定义的 customers 表以及该表的更改事件密钥示例。

表示例

CREATE TABLE customers (
  id SERIAL,
  first_name VARCHAR(255) NOT NULL,
  last_name VARCHAR(255) NOT NULL,
  email VARCHAR(255) NOT NULL,
  PRIMARY KEY(id)
);

更改事件键示例

如果 topic.prefix 连接器配置属性的值为 PostgreSQL_server，则 customer 表的每个更改事件都有相同的键结构，JSON 类似如下：

{
  "schema": { 1
    "type": "struct",
    "name": "PostgreSQL_server.public.customers.Key", 2
    "optional": false, 3
    "fields": [ 4
          {
              "name": "id",
              "index": "0",
              "schema": {
                  "type": "INT32",
                  "optional": "false"
              }
          }
      ]
  },
  "payload": { 5
      "id": "1"
  },
}

表 7.8. 更改事件键的描述
项	字段名称	描述
1	`schema`	键的 schema 部分指定一个 Kafka Connect 模式，它描述了键的 `payload` 部分的内容。
2	`PostgreSQL_server.inventory.customers.Key`	定义密钥有效负载结构的模式名称。这个 schema 描述了已更改的表的主键的结构。键模式名称的格式是 connector-name.database-name.table-name.`Key`.在本例中： `PostgreSQL_server` 是生成此事件的连接器的名称。 `inventory` 是包含已更改表的数据库。 `客户` 是更新的表。
3	`optional`	指明 event 键是否必须在其 `payload` 字段中包含一个值。在本例中，键有效负载中的值是必需的。当表没有主键时，键的 payload 字段中的值是可选的。
4	`fields`	指定 `有效负载中` 预期的每个字段，包括每个字段的名称、索引和模式。
5	`payload`	包含生成此更改事件的行的密钥。在本例中，键包含一个 `id` 字段，其值为 `1`。

注意

虽然 column.exclude.list 和 column.include.list 连接器配置属性允许您只捕获表列的子集，但主键或唯一键中的所有列始终包含在事件键中。

警告

如果表没有主密钥或唯一密钥，则更改事件的密钥为 null。没有主或唯一键约束的表中的行无法唯一标识。

7.3.2. 关于 Debezium PostgreSQL 更改事件中的值

更改事件中的值比键复杂一些。与键一样，该值有一个 schema 部分和 payload 部分。schema 部分包含描述 payload 部分的 Envelope 结构的 schema，包括其嵌套字段。为创建、更新或删除数据的操作更改事件，它们都有一个带有 envelope 结构的值有效负载。

考虑用于显示更改事件键示例的相同示例表：

CREATE TABLE customers (
  id SERIAL,
  first_name VARCHAR(255) NOT NULL,
  last_name VARCHAR(255) NOT NULL,
  email VARCHAR(255) NOT NULL,
  PRIMARY KEY(id)
);

更改此表的更改事件的值部分因 REPLICA IDENTITY 设置和事件所针对的操作而异。

这些部分中的详情如下：

副本身份

REPLICA IDENTITY 是一个特定于 PostgreSQL 的表级设置，它决定了 UPDATE 和 DELETE 事件的逻辑解码插件可用的信息量。更具体地说，当发生 UPDATE 或 DELETE 事件时，会控制 REPLICA IDENTITY 的设置可用于表列的前面值。

REPLICA IDENTITY 有 4 个可能的值：

DEFAULT - 如果该表有主键，则默认行为是 UPDATE 和 DELETE 事件包含表的主键列的前面值。对于 UPDATE 事件，只有带有更改值的主键列才会存在。
如果表没有主密钥，连接器不会为该表发出 UPDATE 或 DELETE 事件。对于没有主密钥的表，连接器只发出 创建事件。通常，没有主键的表用于将消息附加到表的末尾，这意味着 UPDATE 和 DELETE 事件不有用。
NOTHING - 为 UPDATE 和 DELETE 操作传输的事件不包含任何表列的之前值的信息。
FULL - Emitted 事件用于 UPDATE 和 DELETE 操作，包含表中所有列的前面值。
INDEX index-name - Emitted 事件用于 UPDATE 和 DELETE 操作，包含指定索引中包含的列的先前值。UPDATE 事件也包含带有更新值的索引列。

创建事件

以下示例显示了一个更改事件的值部分，连接器为在 customer 表中创建数据的操作生成的更改事件的值部分：

{
    "schema": { 1
        "type": "struct",
        "fields": [
            {
                "type": "struct",
                "fields": [
                    {
                        "type": "int32",
                        "optional": false,
                        "field": "id"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "first_name"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "last_name"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "email"
                    }
                ],
                "optional": true,
                "name": "PostgreSQL_server.inventory.customers.Value", 2
                "field": "before"
            },
            {
                "type": "struct",
                "fields": [
                    {
                        "type": "int32",
                        "optional": false,
                        "field": "id"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "first_name"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "last_name"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "email"
                    }
                ],
                "optional": true,
                "name": "PostgreSQL_server.inventory.customers.Value",
                "field": "after"
            },
            {
                "type": "struct",
                "fields": [
                    {
                        "type": "string",
                        "optional": false,
                        "field": "version"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "connector"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "name"
                    },
                    {
                        "type": "int64",
                        "optional": false,
                        "field": "ts_ms"
                    },
                    {
                        "type": "boolean",
                        "optional": true,
                        "default": false,
                        "field": "snapshot"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "db"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "schema"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "table"
                    },
                    {
                        "type": "int64",
                        "optional": true,
                        "field": "txId"
                    },
                    {
                        "type": "int64",
                        "optional": true,
                        "field": "lsn"
                    },
                    {
                        "type": "int64",
                        "optional": true,
                        "field": "xmin"
                    }
                ],
                "optional": false,
                "name": "io.debezium.connector.postgresql.Source", 3
                "field": "source"
            },
            {
                "type": "string",
                "optional": false,
                "field": "op"
            },
            {
                "type": "int64",
                "optional": true,
                "field": "ts_ms"
            }
        ],
        "optional": false,
        "name": "PostgreSQL_server.inventory.customers.Envelope" 4
    },
    "payload": { 5
        "before": null, 6
        "after": { 7
            "id": 1,
            "first_name": "Anne",
            "last_name": "Kretchmar",
            "email": "annek@noanswer.org"
        },
        "source": { 8
            "version": "2.3.4.Final",
            "connector": "postgresql",
            "name": "PostgreSQL_server",
            "ts_ms": 1559033904863,
            "snapshot": true,
            "db": "postgres",
            "sequence": "[\"24023119\",\"24023128\"]",
            "schema": "public",
            "table": "customers",
            "txId": 555,
            "lsn": 24023128,
            "xmin": null
        },
        "op": "c", 9
        "ts_ms": 1559033904863 10
    }
}

表 7.9. 创建事件值字段的描述
项	字段名称	描述
1	`schema`	值的 schema，用于描述值有效负载的结构。当连接器为特定表生成的每次更改事件中，更改事件的值模式都是相同的。
2	`name`	在 `schema` 部分中，每个 `name` 字段为值有效负载中的字段指定 schema。 `PostgreSQL_server.inventory.customers.Value` 是有效负载 `before` 和 `after` 字段的 schema。这个模式特定于 `customers` 表。 `before` 和 `after` 字段的模式名称格式为 `logicalName.tableName.Value`，这样可确保 schema 名称在数据库中是唯一的。这意味着，在使用 Avro converter 时，每个逻辑源中的每个表生成的 Avro 模式都有自己的 evolution 和 history。
3	`name`	`io.debezium.connector.postgresql.Source` 是有效负载的 `source` 字段的 schema。这个模式特定于 PostgreSQL 连接器。连接器将其用于它生成的所有事件。
4	`name`	`PostgreSQL_server.inventory.customers.Envelope` 是有效负载的整体结构的 schema，其中 `PostgreSQL_server` 是连接器名称，`inventory` 是数据库，`customers` 是表。
5	`payload`	值的实际数据。这是更改事件提供的信息。可能会出现事件的 JSON 表示比它们描述的行大得多。这是因为 JSON 表示必须包含消息的 schema 和 payload 部分。但是，通过使用 Avro converter，您可以显著减少连接器流到 Kafka 主题的信息大小。
6	`before`	指定事件发生前行状态的可选字段。当 `op` 字段是 `c` 用于创建（如本例所示），`before` 字段为 `null`，因为此更改事件用于新内容。注意此字段是否可用取决于每个表的 `REPLICA IDENTITY` 设置。
7	`after`	指定事件发生后行状态的可选字段。在本例中，`after` 字段包含新行的 `id`、`first_name`、`last_name` 和 `email` 列的值。
8	`source`	描述事件源元数据的必需字段。此字段包含可用于将此事件与其他事件进行比较的信息，以及事件的来源、事件发生的顺序以及事件是否为同一事务的一部分。源元数据包括： Debezium 版本连接器类型和名称包含新行的数据库和表专用 JSON 数组的额外偏移信息。第一个值是最后一个提交的 LSN，第二个值是当前的 LSN。两个值可以是 `null`。模式名称如果事件是快照的一部分执行操作的事务的 ID 数据库日志中操作的偏移量在数据库中进行更改时的时间戳
9	`op`	描述导致连接器生成事件的操作类型的强制字符串。在本例中，`c` 表示操作创建了行。有效值为： `c` = create `u` = update `d` = delete `r` = 读取（仅适用于快照） `t` = truncate `m` = message
10	`ts_ms`	可选字段，显示连接器处理事件的时间。这个时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。 `在源` 对象中，`ts_ms` 表示数据库中进行更改的时间。通过将 `payload.source.ts_ms` 的值与 `payload.ts_ms` 的值进行比较，您可以确定源数据库更新和 Debezium 之间的滞后。

更新事件

示例 customers 表中一个更新的改变事件的值有与那个表的 create 事件相同的模式。同样，事件值有效负载具有相同的结构。但是，事件值有效负载在 update 事件中包含不同的值。以下是连接器为 customer 表中更新生成的更改事件值 的示例 ：

{
    "schema": { ... },
    "payload": {
        "before": { 1
            "id": 1
        },
        "after": { 2
            "id": 1,
            "first_name": "Anne Marie",
            "last_name": "Kretchmar",
            "email": "annek@noanswer.org"
        },
        "source": { 3
            "version": "2.3.4.Final",
            "connector": "postgresql",
            "name": "PostgreSQL_server",
            "ts_ms": 1559033904863,
            "snapshot": false,
            "db": "postgres",
            "schema": "public",
            "table": "customers",
            "txId": 556,
            "lsn": 24023128,
            "xmin": null
        },
        "op": "u", 4
        "ts_ms": 1465584025523  5
    }
}

表 7.10. 更新事件值字段的描述
项	字段名称	描述
1	`before`	包含数据库提交前行中的值的可选字段。在此例中只有一个主列 `id`，因为表的 `REPLICA IDENTITY` 设置默认为 `DEFAULT`。 + 对于一个 update 事件来包括行中所有列的以前的值，您需要修改 `customers` 表（通过运行 `ALTER TABLE customers REPLICA IDENTITY FULL`）。
2	`after`	指定事件发生后行状态的可选字段。在本例中，`first_name` 值现在是 `Anne Marie`。
3	`source`	描述事件源元数据的必需字段。`source` 字段结构与 create 事件中的字段相同，但一些值不同。源元数据包括： Debezium 版本连接器类型和名称包含新行的数据库和表模式名称如果事件是快照的一部分（对于更新事件，始终为 `false` ）执行操作的事务的 ID 数据库日志中操作的偏移量在数据库中进行更改时的时间戳
4	`op`	描述操作类型的强制字符串。在 update 事件值中，`op` 字段值为 `u`，表示此行因为更新而改变。
5	`ts_ms`	可选字段，显示连接器处理事件的时间。这个时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。 `在源` 对象中，`ts_ms` 表示数据库中进行更改的时间。通过将 `payload.source.ts_ms` 的值与 `payload.ts_ms` 的值进行比较，您可以确定源数据库更新和 Debezium 之间的滞后。

注意

更新行 primary/unique 键的列会更改行的键值。当键更改时，Debezium 会输出三个事件：一个 DELETE 事件，以及一个带有行的旧键的 tombstone 事件，后跟一个带有行的新键的事件。详情在下一节中。

主密钥更新

更改行的主键字段的 UPDATE 操作称为主密钥更改。对于主键更改，以代替发送 UPDATE 事件记录，连接器会为旧密钥发送 DELETE 事件记录，并为新的(updated)密钥创建一个 CREATE 事件记录。这些事件具有常见的结构和内容，另外，每个事件都有与主密钥更改相关的消息标头：

DELETE 事件记录具有 __debezium.newkey 作为消息标头。此标头的值是更新行的新主键。
CREATE 事件记录具有 __debezium.oldkey 作为消息标头。此标头的值是更新行所具有的前一个主键（旧的）主键。

删除事件

delete 更改事件中的值与为同一表的 create 和 update 事件相同的 schema 部分。示例 customer 表的 delete 事件中 payload 部分类似如下：

{
    "schema": { ... },
    "payload": {
        "before": { 1
            "id": 1
        },
        "after": null, 2
        "source": { 3
            "version": "2.3.4.Final",
            "connector": "postgresql",
            "name": "PostgreSQL_server",
            "ts_ms": 1559033904863,
            "snapshot": false,
            "db": "postgres",
            "schema": "public",
            "table": "customers",
            "txId": 556,
            "lsn": 46523128,
            "xmin": null
        },
        "op": "d", 4
        "ts_ms": 1465581902461 5
    }
}

表 7.11. 删除事件值字段的描述
项	字段名称	描述
1	`before`	指定事件发生前行状态的可选字段。在 delete 事件值中，`before` 字段包含在使用数据库提交删除行前的值。在本例中，`before` 字段仅包含主键列，因为表的 `REPLICA IDENTITY` 设置为 `DEFAULT`。
2	`after`	指定事件发生后行状态的可选字段。在 delete 事件值中，`after` 字段为 `null`，表示行不再存在。
3	`source`	描述事件源元数据的必需字段。在一个 delete 事件值中，`source` 字段结构与同一表的 create 和 update 事件相同。许多 `source` 字段值也相同。在 delete 事件值中，`ts_ms` 和 `lsn` 字段值以及其他值可能已更改。但是 delete 事件值中的 `source` 字段提供相同的元数据： Debezium 版本连接器类型和名称包含已删除行的数据库和表模式名称如果事件是快照的一部分（对于删除事件，始终为 `false` ）执行操作的事务的 ID 数据库日志中操作的偏移量在数据库中进行更改时的时间戳
4	`op`	描述操作类型的强制字符串。`op` 字段值为 `d`，表示此行已被删除。
5	`ts_ms`	可选字段，显示连接器处理事件的时间。这个时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。 `在源` 对象中，`ts_ms` 表示数据库中进行更改的时间。通过将 `payload.source.ts_ms` 的值与 `payload.ts_ms` 的值进行比较，您可以确定源数据库更新和 Debezium 之间的滞后。

删除更改事件记录为消费者提供处理此行删除所需的信息。

警告

要使消费者处理为没有主键的表生成的删除事件，请将表的 REPLICA IDENTITY 设置为 FULL。当表没有主键且表的 REPLICA IDENTITY 设置为 DEFAULT 或 NOTHING 时，删除事件在字段 之前没有。

PostgreSQL 连接器事件旨在使用 Kafka 日志压缩。只要保留每个密钥的最新消息，日志压缩就会启用删除一些旧的消息。这可让 Kafka 回收存储空间，同时确保主题包含完整的数据集，并可用于重新载入基于密钥的状态。

tombstone 事件

删除行时，delete 事件值仍可用于日志压缩，因为 Kafka 您可以删除具有相同键的所有之前信息。但是，要让 Kafka 删除具有相同键的所有消息，消息值必须为 null。为了实现此目的，PostgreSQL 连接器遵循一个 delete 事件，其中包含一个特殊的tombstone 事件，它有相同的键但值为 null。

Truncate 事件

截断更改事件信号，表示表已被截断。在这种情况下，message 键为 null，message 值类似如下：

{
    "schema": { ... },
    "payload": {
        "source": { 1
            "version": "2.3.4.Final",
            "connector": "postgresql",
            "name": "PostgreSQL_server",
            "ts_ms": 1559033904863,
            "snapshot": false,
            "db": "postgres",
            "schema": "public",
            "table": "customers",
            "txId": 556,
            "lsn": 46523128,
            "xmin": null
        },
        "op": "t", 2
        "ts_ms": 1559033904961 3
    }
}

表 7.12. truncate 事件值字段的描述
项	字段名称	描述
1	`source`	描述事件源元数据的必需字段。在 truncate 事件值中，`source` 字段结构与为同一表的 create, update, 和 delete 事件相同，提供此元数据： Debezium 版本连接器类型和名称包含新行的数据库和表模式名称如果事件是快照的一部分（对于删除事件，始终为 `false` ）执行操作的事务的 ID 数据库日志中操作的偏移量在数据库中进行更改时的时间戳
2	`op`	描述操作类型的强制字符串。`op` 字段值为 `t`，表示此表已被截断。
3	`ts_ms`	可选字段，显示连接器处理事件的时间。这个时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。 `在源` 对象中，`ts_ms` 表示数据库中进行更改的时间。通过将 `payload.source.ts_ms` 的值与 `payload.ts_ms` 的值进行比较，您可以确定源数据库更新和 Debezium 之间的滞后。

如果单个 TRUNCATE 语句应用到多个表，则会为每个删节的表发出一个 truncate 更改事件记录。

请注意，由于 truncate 事件代表对整个表进行的更改，且没有消息密钥，除非您使用单个分区的主题，否则与表相关的更改事件没有排序保证(创建、更新等)和截断该表的事件。例如，当这些事件从不同的分区读取时，消费者只能在该表的 truncate 事件后收到 update 事件。

消息事件

只有 Postgres 14+ 上的 pgoutput 插件支持此事件类型(Postgres 文档)

消息事件信号，一个通用逻辑解码消息已被直接插入到 WAL 中，通常带有 pg_logical_emit_message 函数。message 键是一个 Struct，其中包含一个名为 prefix 的单个字段，它会执行插入消息时指定的前缀。message 值与事务性信息类似：

{
    "schema": { ... },
    "payload": {
        "source": { 1
            "version": "2.3.4.Final",
            "connector": "postgresql",
            "name": "PostgreSQL_server",
            "ts_ms": 1559033904863,
            "snapshot": false,
            "db": "postgres",
            "schema": "",
            "table": "",
            "txId": 556,
            "lsn": 46523128,
            "xmin": null
        },
        "op": "m", 2
        "ts_ms": 1559033904961, 3
        "message": { 4
            "prefix": "foo",
            "content": "Ymfy"
        }
    }
}

与其他事件类型不同，非事务性消息将没有任何关联的 BEGIN 或 END 事务事件。对于非事务信息，message 值类似如下：

{
    "schema": { ... },
    "payload": {
        "source": { 1
            "version": "2.3.4.Final",
            "connector": "postgresql",
            "name": "PostgreSQL_server",
            "ts_ms": 1559033904863,
            "snapshot": false,
            "db": "postgres",
            "schema": "",
            "table": "",
            "lsn": 46523128,
            "xmin": null
        },
        "op": "m", 2
        "ts_ms": 1559033904961 3
        "message": { 4
            "prefix": "foo",
            "content": "Ymfy"
    }
}

表 7.13. 消息事件值字段的描述
项	字段名称	描述
1	`source`	描述事件源元数据的必需字段。在 message 事件值中，`source` 字段结构将不会包括任何 message 事件的 `table` 或 `schema` 信息，并只有在 message 事件是事务性时才有 `txId`。 Debezium 版本连接器类型和名称数据库名称 schema name (always `""` for message events) 表名称（始终为 `""` 消息事件）如果事件是快照的一部分（对于消息事件，始终为 `false` ）执行操作的事务的 ID （非事务 message 事件为 `null`）数据库日志中操作的偏移量事务消息：当消息插入到 WAL 中时的时间戳非交易消息；当连接器遇到消息时的 Timestamp
2	`op`	描述操作类型的强制字符串。`op` 字段值为 `m`，表示这是消息事件。
3	`ts_ms`	可选字段，显示连接器处理事件的时间。这个时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。对于事务消息事件，源对象的 `ts_ms` 属性指示数据库中更改的时间。通过将 `payload.source.ts_ms` 的值与 `payload.ts_ms` 的值进行比较，您可以确定源数据库更新和 Debezium 之间的滞后。对于非事务消息事件，`源对象的` `ts_ms` 表示连接器遇到消息事件的时间，而 `payload.ts_ms` 表示连接器处理事件的时间。这是因为，提交时间戳在 Postgres 的通用逻辑消息格式中不存在，且非事务逻辑消息没有以 `BEGIN` 事件（具有时间戳信息）的前面。
4	`message`	包含消息元数据的字段前缀（文本）内容（基于二进制处理模式设置编码的字节阵列）

7.3. Debezium PostgreSQL 连接器数据更改事件的描述

7.3.1. 关于 Debezium PostgreSQL 中的键更改事件

7.3.2. 关于 Debezium PostgreSQL 更改事件中的值

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Red Hat legal and privacy links

Red Hat legal and privacy links