MySQL CDC 实战：Debezium 变更数据格式详解

原创

叫我阿柒啊

发布于 2025-07-18 23:56:19

2260

文章被收录于专栏：大数据之路大数据之路入门到放弃之路

前言

在上一篇Debezium 踩坑实录：我在MySQL CDC 遇到的3个问题. 文章中，讲述了我在 debezium 程序中遇到的问题和解决方法，本篇文章就按照主线任务继续往下走，来研究一下如何处理捕获的变更数据。

程序开发

我们通过之前的调试，以及根据自己的需求完善了配置，保证程序只采集 insert、update、delete 中 dml 操作的变更数据，完整配置如下：

final Properties props = new Properties();
props.setProperty("name", "engine");
props.setProperty("connector.class", "io.debezium.connector.mysql.MySqlConnector");
props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
props.setProperty("offset.storage.file.filename", "offsets.dat");
props.setProperty("offset.flush.interval.ms", "60000");
/* begin connector properties */
props.setProperty("database.hostname", "cvm");
props.setProperty("database.port", "3306");
props.setProperty("database.user", "root");
props.setProperty("database.password", "123456");
props.setProperty("database.server.id", "85744");
props.setProperty("table.include.list", "debezium.test");
props.setProperty("topic.prefix", "aqi");
props.setProperty("schema.history.internal", "io.debezium.storage.file.history.FileSchemaHistory");
props.setProperty("schema.history.internal.file.filename", "schemahistory.dat");
props.setProperty("converter.schemas.enable", "false");
props.setProperty("include.schema.changes", "false");

数据处理

接下来就开始关注我们的 notifying 数据处理的部分。

    try (DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
            .using(props)
            .notifying(record -> {
                System.out.println(record);
            }).build()
    ) {
        // Run the engine asynchronously ...
        ExecutorService executor = Executors.newSingleThreadExecutor();
        executor.execute(engine);

        for (;;) {}
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
  }
}

那么如何解析采集到的数据呢，关键就在于 record，这是一个 EmbeddedEngineChangeEvent<String, String> 对象，我们可以看一下关于这个类的定义。

这个类一共有四个成员变量，其中我们只需要关注 key 和 value 即可，key 可以理解为采集到变更数据的唯一表示，value 就是采集到的数据变更信息。然后我在 nitifying 打印出来（这种内部函数断点是真不好加）。

可以看到 key 和 value 都是 json，record 的 headers 和 partition 都为空，因为这里不是 Kafka Connect 模式，最后的 destination 是 Kafka Connect 模式下，自动生成的 topic 信息，这里也是印证了 topic 生成规则是：topicPrefix.databaseName.tableName。

json格式化

key 的话就一个 id 字段，这里可以先不关注，我们只关注 value 即可，我们复制 value 格式化一下 json，查看其中的字段。

其中json的一级节点是 before、after、source、transaction、op 以及三个类似于时间戳的字段，我觉得 op 字段枚举值比较关键，因为通过 op 我们才能知道执行了什么类型的 sql：

c = create
u = update
d = delete
r = read (applies to only snapshots)

上面是官方文档给出的解释，如上图所示是我执行 insert 语句捕获到的数据，所以 op 为 c，表示新插入的数据。

然后就是 before 和 after，before 是表示变更前的数据，因为 insert 是新插入的数据，所以 before 就为 null，如果是 update 语句的话，before 就是有值的。after 就是变更后的数据，这里包含表中的字段。

至于 source，打眼一看就是之前配置上的东西，包括 debezium 的配置、database、table 的配置等等。如果你看了官网文档，你应该会发现应该还会输出一个 schema 节点，但是这里明显没有输出，因为我在配置中，设置 converter.schemas.enable 为 "false"，这样就不会打印表的schema信息了。