第 4 章 查看更改事件

流程

1. 打开一个新终端，并使用 kafka-console-consumer 来消耗主题开头的 dbserver1.inventory.customers 主题。

   此命令在运行 Kafka (my-cluster-kafka-0)的 Pod 中运行一个简单的消费者(kafka-console-consumer.sh)：

   $ oc exec -it my-cluster-kafka-0 -- /opt/kafka/bin/kafka-console-consumer.sh \
     --bootstrap-server localhost:9092 \
     --from-beginning \
     --property print.key=true \
     --topic dbserver1.inventory.customers

   Copy to Clipboard Toggle word wrap

   消费者返回四个消息（以 JSON 格式），在 customers 表中每行都对应一个信息。每个消息都包含对应表行的事件记录。

   每个事件有两个 JSON 文档：一个 key 和一个 value。键对应于行的主键，而值则显示行的详细信息（行包含的字段、每个字段的值以及行执行的操作类型）。

2. 对于最后一个事件，请查看 密钥 的详细信息。

   以下是最后一次事件 的密钥 的详细信息（为可读性格式化）：

   {
     "schema":{
       "type":"struct",
         "fields":[
           {
             "type":"int32",
             "optional":false,
             "field":"id"
           }
         ],
       "optional":false,
       "name":"dbserver1.inventory.customers.Key"
     },
     "payload":{
       "id":1004
     }
   }

   Copy to Clipboard Toggle word wrap

   事件有两个部分： schema 和 payload。模式 包含一个 Kafka Connect 模式，描述有效负载中的内容。在这种情况下，有效负载是名为 dbserver1.inventory.customers.Key 的 struct，它是可选的，且有一个必填字段（类型为 int32 的 id ）。

   payload 是一个单一的 id 字段，它的值为 1004。

   通过查看事件的 key，您可以看到此事件应用到 inventory.customers 表中的 id 主键栏的值为 1004 的行。

3. 查看同一事件 值 的详细信息。

   事件 的值 显示行已创建，并描述了它包含的内容（本例中为 id、first_name、last_name 以及插入行 的电子邮件 ）。

   以下是最后一次事件 的值 的详细信息（格式为可读性）：

   {
     "schema": {
       "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": "dbserver1.inventory.customers.Value",
           "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": "dbserver1.inventory.customers.Value",
           "field": "after"
         },
         {
           "type": "struct",
           "fields": [
             {
               "type": "string",
               "optional": true,
               "field": "version"
             },
             {
               "type": "string",
               "optional": false,
               "field": "name"
             },
             {
               "type": "int64",
               "optional": false,
               "field": "server_id"
             },
             {
               "type": "int64",
               "optional": false,
               "field": "ts_sec"
             },
             {
               "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": "boolean",
               "optional": true,
               "field": "snapshot"
             },
             {
               "type": "int64",
               "optional": true,
               "field": "thread"
             },
             {
               "type": "string",
               "optional": true,
               "field": "db"
             },
             {
               "type": "string",
               "optional": true,
               "field": "table"
             }
           ],
           "optional": false,
           "name": "io.debezium.connector.mysql.Source",
           "field": "source"
         },
         {
           "type": "string",
           "optional": false,
           "field": "op"
         },
         {
           "type": "int64",
           "optional": true,
           "field": "ts_ms"
         },
         {
           "type": "int64",
           "optional": true,
           "field": "ts_us"
         },
         {
           "type": "int64",
           "optional": true,
           "field": "ts_ns"
         }
       ],
       "optional": false,
       "name": "dbserver1.inventory.customers.Envelope",
       "version": 1
     },
     "payload": {
       "before": null,
       "after": {
         "id": 1004,
         "first_name": "Anne",
         "last_name": "Kretchmar",
         "email": "annek@noanswer.org"
       },
       "source": {
         "version": "3.0.8.Final",
         "name": "dbserver1",
         "server_id": 0,
         "ts_sec": 0,
         "gtid": null,
         "file": "mysql-bin.000003",
         "pos": 154,
         "row": 0,
         "snapshot": true,
         "thread": null,
         "db": "inventory",
         "table": "customers"
       },
       "op": "r",
       "ts_ms": 1486500577691,
       "ts_us": 1486500577691547,
       "ts_ns": 1486500577691547930
     }
   }

   Copy to Clipboard Toggle word wrap

   事件的这一部分时间较长，但与事件 的密钥 一样，它还具有 schema 和 payload。schema 包括了一个 Kafka Connect 方案，称为 dbserver1.inventory.customers.Envelope (version 1)，它可以包括五个字段：

   op

   包含描述操作类型的字符串值的必填字段。MySQL 连接器的值是 c 用于创建（或插入）、u 用于 update，d 表示删除，r 用于读取（在快照的情况下）。

   before

   存在的可选字段（如果存在）包含事件 发生前 行的状态。该结构由 dbserver1.inventory.customers.Value Kafka Connect 模式描述，dbserver1 连接器用于 inventory.customers 表中的所有行。

   after

   存在的可选字段（如果存在）包含事件发生 后 行的状态。该结构与之前使用的 dbserver1.inventory.customers.Value Kafka Connect schema 相同。

   source

   一个必需的字段，包含描述事件源元数据的结构（在 MySQL 中，包含几个字段：连接器名称、记录事件的 binlog 文件的名称、事件出现的 binlog 的位置，事件中显示的位置（如果有多个字段）、受影响的数据库和表的名称，进行更改的 MySQL 线程 ID，这个事件是快照的一部分，如果可用，则 MySQL 服务器 ID 和时间戳（以秒为单位）。

   ts_ms

   可选字段，如果存在，包含运行 Kafka Connect 任务的 JVM 中的时间（使用系统时钟），该字段处理事件。

   注意

   事件的 JSON 表示比它们描述的行要长。这是因为，在每个事件键和值中，Kafka Connect 都附带了描述 有效负载的 schema。随着时间的推移，这种结构可能会改变。但是，在事件本身中拥有 key 和 value 的 schema 可让您更轻松地使用应用程序了解消息，特别是随着时间推移而变化的信息。

   Debezium MySQL 连接器根据数据库表的结构构建这些模式。如果您使用 DDL 语句更改 MySQL 数据库中的表定义，连接器会读取这些 DDL 语句并更新其 Kafka Connect 模式。这是每个事件的结构与事件发生时源自的表的唯一方式。但是，包含单个表的所有事件的 Kafka 主题可能会具有与表定义的每个状态对应的事件。

   JSON 转换程序在每个消息中包含键和值模式，因此它会产生非常详细的事件。

4. 将事件的 key 和 value 方案与 inventory 数据库的状态进行比较。在运行 MySQL 命令行客户端的终端中，运行以下语句：

   mysql> SELECT * FROM customers;
   +------+------------+-----------+-----------------------+
   | id   | first_name | last_name | email                 |
   +------+------------+-----------+-----------------------+
   | 1001 | Sally      | Thomas    | sally.thomas@acme.com |
   | 1002 | George     | Bailey    | gbailey@foobar.com    |
   | 1003 | Edward     | Walker    | ed@walker.com         |
   | 1004 | Anne       | Kretchmar | annek@noanswer.org    |
   +------+------------+-----------+-----------------------+
   4 rows in set (0.00 sec)

   Copy to Clipboard Toggle word wrap

   这表明您检查的事件记录与数据库中记录匹配的记录。