Debezium日常分享系列之：Debezium2.5稳定版本之Mysql连接器的数据变更事件

一、数据变更事件

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

Debezium 和 Kafka Connect 是围绕连续的事件消息流而设计的。然而，这些事件的结构可能会随着时间的推移而改变，这对消费者来说可能很难处理。为了解决这个问题，每个事件都包含其内容的架构，或者，如果您使用架构注册表，则还包含消费者可用于从注册表获取架构的架构 ID。这使得每个事件都是独立的。

以下 JSON 框架显示了更改事件的基本四个部分。但是，您选择在应用程序中使用的 Kafka Connect 转换器的配置方式决定了这四个部分在更改事件中的表示。仅当您配置转换器来生成模式字段时，模式字段才会处于更改事件中。同样，仅当您配置转换器来生成事件键和事件负载时，事件键和事件负载才会出现在更改事件中。如果您使用 JSON 转换器并将其配置为生成所有四个基本更改事件部分，则更改事件具有以下结构：

{
 "schema": { 1
   ...
  },
 "payload": { 2
   ...
 },
 "schema": { 3
   ...
 },
 "payload": { 4
   ...
 },
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

表 7. 变更事件基本内容概述

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

注意：

• MySQL 连接器确保所有 Kafka Connect 架构名称都遵循 Avro 架构名称格式。这意味着逻辑服务器名称必须以拉丁字母或下划线开头，即 a-z、A-Z 或 _。逻辑服务器名称中的每个剩余字符以及数据库和表名称中的每个字符都必须是拉丁字母、数字或下划线，即 a-z、A-Z、0-9 或 _。如果存在无效字符，则将其替换为下划线字符。
• 如果逻辑服务器名称、数据库名称或表名称包含无效字符，并且唯一区分名称的字符无效并用下划线替换，则可能会导致意外冲突。

二、更改事件键

更改事件的键包含已更改表的键和已更改行的实际键的架构。连接器创建事件时，架构及其相应的负载都包含已更改表的主键（或唯一约束）中每一列的字段。

考虑以下客户表，后面是该表的更改事件键的示例。

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;
1
2
3
4
5
6

捕获客户表更改的每个更改事件都具有相同的事件键架构。只要客户表具有先前的定义，捕获客户表更改的每个更改事件都具有以下键结构。在 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
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

表8. 变更事件键说明

三、更改事件值

更改事件中的值比键稍微复杂一些。与键一样，值也具有模式部分和有效负载部分。模式部分包含描述有效负载部分的 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;
1
2
3
4
5
6

对此表的更改的更改事件的值部分描述为：

• 创建事件
• 更新事件
• 主键更新
• 删除事件
• 墓碑事件
• 截断事件

四、创建事件

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

{
  "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.5.3.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')"
    }
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181

表 9. 创建事件值字段的描述

五、更新事件

示例客户表中更新的更改事件的值与该表的创建事件具有相同的架构。同样，事件值的有效负载具有相同的结构。但是，事件值有效负载在更新事件中包含不同的值。以下是连接器为更新客户表而生成的事件中的更改事件值的示例：

{
  "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.5.3.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）
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

表 10. 更新事件值字段说明

注意：

• 更新行的主/唯一键的列会更改行的键的值。当键更改时，Debezium 输出三个事件：DELETE 事件和带有该行旧键的逻辑删除事件，然后是带有该行新键的事件。详细信息在下一节中。

六、主键更新

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

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

七、删除事件

删除更改事件中的值与同一表的创建和更新事件具有相同的架构部分。示例客户表的删除事件中的有效负载部分如下所示：

{
  "schema": { ... },
  "payload": {
    "before": { （1）
      "id": 1004,
      "first_name": "Anne Marie",
      "last_name": "Kretchmar",
      "email": "annek@noanswer.org"
    },
    "after": null, （2）
    "source": { （3）
      "version": "2.5.3.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）
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

表 11. 删除事件值字段说明

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

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

八、墓碑事件

当删除一行时，删除事件值仍然适用于日志压缩，因为 Kafka 可以删除具有相同键的所有早期消息。但是，为了让 Kafka 删除具有相同键的所有消息，消息值必须为 null。为了实现这一点，在 Debezium 的 MySQL 连接器发出删除事件后，连接器会发出一个特殊的逻辑删除事件，该事件具有相同的键但为空值。

九、截断事件

截断更改事件表示表已被截断。在这种情况下，消息键为 null，消息值如下所示：

{
    "schema": { ... },
    "payload": {
        "source": { （1）
            "version": "2.5.3.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）
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

表 12. 截断事件值字段的描述

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

请注意，由于截断事件表示对整个表所做的更改并且没有消息键，因此除非您正在处理具有单个分区的主题，否则与表相关的更改事件没有顺序保证（创建、更新等）并截断该表的事件。例如，当从不同分区读取这些事件时，消费者仅在该表的截断事件之后才可以接收更新事件。

十、Debezium技术总结

更多Debezium技术请参考：

• Debezium技术专栏