5.2. Debezium MySQL 连接器数据更改事件的描述

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

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

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

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

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

默认情况下，连接器流将事件记录更改为名称与事件原始表相同的名称的主题。请参阅主题名称。

警告

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

如果逻辑服务器名称、数据库名称或表名称包含无效字符，且区分另一个名称的唯一字符无效，则可能会导致意外冲突，因此替换为下划线。

更多详情包括在以下主题中：

5.2.1. 关于 Debezium MySQL 中的键更改事件

更改事件的密钥包含更改表的密钥的 schema，以及更改的行的实际键。在连接器创建事件时，模式及其对应有效负载都包含更改表的 PRIMARY KEY （或唯一约束）中的每个列的字段。

请考虑以下 客户 表，然后是此表的更改事件键示例。

CREATE TABLE customers (
  id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
  first_name VARCHAR(255) NOT NULL,
  last_name VARCHAR(255) NOT NULL,
  email VARCHAR(255) NOT NULL UNIQUE KEY
) AUTO_INCREMENT=1001;

捕获对 customers 表的更改的每个更改事件都有相同的事件键模式。只要 customers 表有以前的定义，捕获 customer 表更改的事件都有以下关键结构：在 JSON 中，它类似如下：

{
 "schema": { 1
    "type": "struct",
    "name": "mysql-server-1.inventory.customers.Key", 2
    "optional": false, 3
    "fields": [ 4
      {
        "field": "id",
        "type": "int32",
        "optional": false
      }
    ]
  },
 "payload": { 5
    "id": 1001
  }
}

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

5.2.2. 关于 Debezium MySQL 更改事件中的值

更改事件中的值比键稍微复杂。与键一样，该值具有 schema 部分和 payload 部分。schema 部分包含描述 payload 部分的 Envelope 结构的 schema，包括其嵌套字段。更改创建、更新或删除数据的事件，并且具有 envelope 结构的值有效负载。

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

CREATE TABLE customers (
  id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
  first_name VARCHAR(255) NOT NULL,
  last_name VARCHAR(255) NOT NULL,
  email VARCHAR(255) NOT NULL UNIQUE KEY
) AUTO_INCREMENT=1001;

有关此表的更改事件的值部分，如下所述：

创建事件

以下示例显示了连接器为在 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": "mysql-server-1.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": "mysql-server-1.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": true,
            "field": "table"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "server_id"
          },
          {
            "type": "string",
            "optional": true,
            "field": "gtid"
          },
          {
            "type": "string",
            "optional": false,
            "field": "file"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "pos"
          },
          {
            "type": "int32",
            "optional": false,
            "field": "row"
          },
          {
            "type": "int64",
            "optional": true,
            "field": "thread"
          },
          {
            "type": "string",
            "optional": true,
            "field": "query"
          }
        ],
        "optional": false,
        "name": "io.debezium.connector.mysql.Source", 3
        "field": "source"
      },
      {
        "type": "string",
        "optional": false,
        "field": "op"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ms"
      }
    ],
    "optional": false,
    "name": "mysql-server-1.inventory.customers.Envelope" 4
  },
  "payload": { 5
    "op": "c", 6
    "ts_ms": 1465491411815, 7
    "before": null, 8
    "after": { 9
      "id": 1004,
      "first_name": "Anne",
      "last_name": "Kretchmar",
      "email": "annek@noanswer.org"
    },
    "source": { 10
      "version": "2.1.4.Final",
      "connector": "mysql",
      "name": "mysql-server-1",
      "ts_ms": 0,
      "snapshot": false,
      "db": "inventory",
      "table": "customers",
      "server_id": 0,
      "gtid": null,
      "file": "mysql-bin.000003",
      "pos": 154,
      "row": 0,
      "thread": 7,
      "query": "INSERT INTO customers (first_name, last_name, email) VALUES ('Anne', 'Kretchmar', 'annek@noanswer.org')"
    }
  }
}

表 5.9. 创建事件值字段的描述
项	字段名称	描述
1	`schema`	值的 schema，用于描述值有效负载的结构。每次连接器为特定表生成的更改时，更改事件的值模式都是相同的。
2	`name`	在 `schema` 部分中，每个 `name` 字段指定值有效负载中字段的 schema。 `mysql-server-1.inventory.customers.Value` 是有效负载 `before` 和 `after` 字段的 schema。这个模式特定于 `customers` 表。 `before` 和 `after` 字段的 schema 名称格式为 `logicalName。tableName.Value`，这样可确保架构名称在数据库中是唯一的。这意味着，在使用 Avro converter 时，每个逻辑源中的每个表生成的 Avro 模式都有自己的演进和历史记录。
3	`name`	`io.debezium.connector.mysql.Source` 是有效负载的 `source` 字段的 schema。这个模式特定于 MySQL 连接器。连接器将其用于它生成的所有事件。
4	`name`	`mysql-server-1.inventory.customers.Envelope` 是负载总体结构的模式，其中 `dbserver1` 是连接器名称，`inventory` 是数据库，`customers` 是表。
5	`payload`	值的实际数据。这是更改事件提供的信息。似乎，事件的 JSON 表示要大于它们描述的行。这是因为 JSON 表示必须包含消息的 schema 和 payload 部分。但是，通过使用 Avro converter，您可以显著减少连接器流到 Kafka 主题的信息大小。
6	`op`	描述导致连接器生成事件的操作类型的必要字符串。在本例中，`c` 表示操作创建了行。有效值为： `c` = create `u` = update `d` = delete `r` = read （仅适用于快照）
7	`ts_ms`	显示连接器处理事件的时间的可选字段。该时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。 `在源` 对象中 `，ts_ms` 表示数据库中更改的时间。通过将 `payload.source.ts_ms` 的值与 `payload.ts_ms` 的值进行比较，您可以确定源数据库更新和 Debezium 之间的滞后。
8	`之前`	指定事件发生前行状态的可选字段。当 `op` 字段是 `c` 用于创建（如本例所示），`before` 字段为 `null`，因为此更改事件用于新内容。
9	`after`	指定事件发生后行状态的可选字段。在本例中，`after` 字段包含新行的 `id`、`first_name`、`last_name` 和 `email` 列的值。
10	`source`	描述事件源元数据的必需字段。此字段包含可用于将此事件与其他事件进行比较的信息，以及事件的来源、事件发生的顺序，以及事件是否为同一事务的一部分。源元数据包括： Debezium 版本连接器名称记录事件的 binlog 名称 binlog 位置事件中的行如果事件是快照的一部分包含新行的数据库和表的名称创建事件的 MySQL 线程 ID （仅限非快照） MySQL 服务器 ID （如果可用）在数据库中进行更改的时间戳如果启用了 `binlog_rows_query_log_events` MySQL 配置选项，并且启用了连接器配置 `include.query` 属性，`source` 字段还提供 `查询` 字段，其中包含导致更改事件的原始 SQL 语句。

更新事件

示例 customer 表中更新的更改事件值与该表的 create 事件相同。同样，事件的 payload 具有相同的结构。但是，事件值 payload 在更新 事件中包含不同的值。以下是连接器在 customers 表中为更新生成的更改事件值示例：

{
  "schema": { ... },
  "payload": {
    "before": { 1
      "id": 1004,
      "first_name": "Anne",
      "last_name": "Kretchmar",
      "email": "annek@noanswer.org"
    },
    "after": { 2
      "id": 1004,
      "first_name": "Anne Marie",
      "last_name": "Kretchmar",
      "email": "annek@noanswer.org"
    },
    "source": { 3
      "version": "2.1.4.Final",
      "name": "mysql-server-1",
      "connector": "mysql",
      "name": "mysql-server-1",
      "ts_ms": 1465581029100,
      "snapshot": false,
      "db": "inventory",
      "table": "customers",
      "server_id": 223344,
      "gtid": null,
      "file": "mysql-bin.000003",
      "pos": 484,
      "row": 0,
      "thread": 7,
      "query": "UPDATE customers SET first_name='Anne Marie' WHERE id=1004"
    },
    "op": "u", 4
    "ts_ms": 1465581029523 5
  }
}

表 5.10. 更新事件值字段的描述
项	字段名称	描述
1	`之前`	指定事件发生前行状态的可选字段。在 update 事件值中，`before` 字段包含每个表列的一个字段，以及在数据库提交前在该列中的值。在本例中，`first_name` 值为 `Anne。`
2	`after`	指定事件发生后行状态的可选字段。您可以比较 `before` 和 `after` 结构，以确定此行的更新是什么。在示例中，`first_name` 值现在是 `Anne Marie`。
3	`source`	描述事件源元数据的必需字段。`source` 字段结构与 create 事件中的字段相同，但一些值有所不同，例如，示例 update 事件来自 binlog 中的不同位置。源元数据包括： Debezium 版本连接器名称记录事件的 binlog 名称 binlog 位置事件中的行如果事件是快照的一部分包含更新行的数据库和表的名称创建事件的 MySQL 线程 ID （仅限非快照） MySQL 服务器 ID （如果可用）在数据库中进行更改的时间戳如果启用了 `binlog_rows_query_log_events` MySQL 配置选项，并且启用了连接器配置 `include.query` 属性，`source` 字段还提供 `查询` 字段，其中包含导致更改事件的原始 SQL 语句。
4	`op`	描述操作类型的强制字符串。在 update 事件值中，`op` 字段值为 `u`，表示此行因为更新而改变。
5	`ts_ms`	显示连接器处理事件的时间的可选字段。该时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。 `在源` 对象中 `，ts_ms` 表示数据库中更改的时间。通过将 `payload.source.ts_ms` 的值与 `payload.ts_ms` 的值进行比较，您可以确定源数据库更新和 Debezium 之间的滞后。

注意

更新行的主/唯一键的列会更改行键的值。当键更改时，Debebe 会输出三个事件： DELETE 事件和带有行的旧键的 tombstone 事件，后跟带有行的新键的事件。下一小节中提供了详细信息。

主密钥更新

更改行主密钥字段的 UPDATE 操作被称为主密钥更改。对于主键更改，为了代替 UPDATE 事件记录，连接器会为旧密钥和新（更新）键的 CREATE 事件记录发出 DELETE 事件记录。这些事件具有常见的结构和内容，以及每个事件都有一个与主密钥更改相关的消息标头：

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

删除事件

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

{
  "schema": { ... },
  "payload": {
    "before": { 1
      "id": 1004,
      "first_name": "Anne Marie",
      "last_name": "Kretchmar",
      "email": "annek@noanswer.org"
    },
    "after": null, 2
    "source": { 3
      "version": "2.1.4.Final",
      "connector": "mysql",
      "name": "mysql-server-1",
      "ts_ms": 1465581902300,
      "snapshot": false,
      "db": "inventory",
      "table": "customers",
      "server_id": 223344,
      "gtid": null,
      "file": "mysql-bin.000003",
      "pos": 805,
      "row": 0,
      "thread": 7,
      "query": "DELETE FROM customers WHERE id=1004"
    },
    "op": "d", 4
    "ts_ms": 1465581902461 5
  }
}

表 5.11. 删除事件值字段的描述
项	字段名称	描述
1	`之前`	指定事件发生前行状态的可选字段。在一个 delete 事件值中，`before` 字段包含在使用数据库提交删除行前的值。
2	`after`	指定事件发生后行状态的可选字段。在 delete 事件值中，`after` 字段为 `null`，表示行不再存在。
3	`source`	描述事件源元数据的必需字段。在一个 delete 事件值中，`source` 字段结构与同一表的 create 和 update 事件相同。许多 `source` 字段值也相同。在 delete 事件值中，`ts_ms` 和 `pos` 字段值以及其他值可能已更改。但是，delete 事件值中的 `source` 字段提供了相同的元数据： Debezium 版本连接器名称记录事件的 binlog 名称 binlog 位置事件中的行如果事件是快照的一部分包含更新行的数据库和表的名称创建事件的 MySQL 线程 ID （仅限非快照） MySQL 服务器 ID （如果可用）在数据库中进行更改的时间戳如果启用了 `binlog_rows_query_log_events` MySQL 配置选项，并且启用了连接器配置 `include.query` 属性，`source` 字段还提供 `查询` 字段，其中包含导致更改事件的原始 SQL 语句。
4	`op`	描述操作类型的强制字符串。`op` 字段值为 `d`，表示此行已被删除。
5	`ts_ms`	显示连接器处理事件的时间的可选字段。该时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。 `在源` 对象中 `，ts_ms` 表示数据库中更改的时间。通过将 `payload.source.ts_ms` 的值与 `payload.ts_ms` 的值进行比较，您可以确定源数据库更新和 Debezium 之间的滞后。

删除更改事件记录为消费者提供处理此行删除所需的信息。包括了旧值，因为有些使用者可能需要它们才能正确处理删除。

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

tombstone 事件

删除行时，delete 事件值仍可用于日志压缩，因为 Kafka 您可以删除具有相同键的所有之前信息。但是，如果 Kafka 删除具有相同键的所有消息，消息值必须是 null。为了实现此目的，在 Debezium 的 MySQL 连接器发出一个 delete 事件后，连接器会发出一个特殊的 tombstone 事件，它具有相同的键但有一个 null 值。

截断事件

一个截断的更改事件信号，表示表已被截断。本例中 message 键为 null，消息值类似如下：

{
    "schema": { ... },
    "payload": {
        "source": { 1
            "version": "2.1.4.Final",
            "name": "mysql-server-1",
            "connector": "mysql",
            "name": "mysql-server-1",
            "ts_ms": 1465581029100,
            "snapshot": false,
            "db": "inventory",
            "table": "customers",
            "server_id": 223344,
            "gtid": null,
            "file": "mysql-bin.000003",
            "pos": 484,
            "row": 0,
            "thread": 7,
            "query": "UPDATE customers SET first_name='Anne Marie' WHERE id=1004"
        },
        "op": "t", 2
        "ts_ms": 1465581029523 3
    }
}

表 5.12. 截断事件值字段的描述
项	字段名称	描述
1	`source`	描述事件源元数据的必需字段。在 truncate 事件值中，`source` 字段结构与为同一表创建、更新和删除事件相同，提供此元数据： Debezium 版本连接器类型和名称记录事件的 binlog 名称 binlog 位置事件中的行如果事件是快照的一部分数据库和表的名称截断事件的 MySQL 线程 ID （仅限非快照） MySQL 服务器 ID （如果可用）在数据库中进行更改的时间戳
2	`op`	描述操作类型的强制字符串。`op` 字段值为 `t`，表示此表已被截断。
3	`ts_ms`	显示连接器处理事件的时间的可选字段。该时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。 + `在源` 对象中，`ts_ms` 表示数据库中更改的时间。通过将 `payload.source.ts_ms` 的值与 `payload.ts_ms` 的值进行比较，您可以确定源数据库更新和 Debezium 之间的滞后。

如果单个 TRUNCATE 语句应用到多个表，则会发出每个截断表的截断更改事件记录。

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

5.2. Debezium MySQL 连接器数据更改事件的描述

5.2.1. 关于 Debezium MySQL 中的键更改事件

5.2.2. 关于 Debezium MySQL 更改事件中的值

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Red Hat legal and privacy links

Red Hat legal and privacy links