Debezium 实战：使用 Gson 解析 CDC 变更数据

原创

叫我阿柒啊

发布于 2025-07-20 23:54:12

3130

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

前言

在上一篇MySQL CDC 实战：Debezium 变更数据格式详解文章中，探究了一番 debezium 采集到的变更数据格式。在原始格式的基础上，我们可以只去其中的部分有效字段，构建成一个新的 json 数据写入到 Kafka 中。

解析

首先我们需要解析采集到的变更数据，这里我选择 gson 解析数据，在 maven 中我引入 gson 依赖：

<dependency>
    <groupId>com.google.code.gson</groupId>
    <artifactId>gson</artifactId>
    <version>2.10.1</version>
</dependency>

然后我们回顾一下之前采集到的数据格式，明确一下数据字段。

然后根据上面的字段和层级，我们定义实体类来完成字段映射：

public class DebeziumEvent {
    public RowData before;
    public RowData after;
    public Source source;
    public String op;
    public long ts_ms;
    public long ts_us;
    public long ts_ns;

    static class RowData {
        public int id;
        public String name;
        public int age;

        @SerializedName("updated_at")
        public String updatedAt;
    }

    static class Source {
        public String version;
        public String connector;
        public String name;
        public long ts_ms;
        public String snapshot;
        public String db;
        public String sequence;
        public long ts_us;
        public long ts_ns;
        public String table;
        public long server_id;
        public String gtid;
        public String file;
        public long pos;
        public int row;
        public long thread;
        public String query;
    }

我们定义了 DebeziumEvent 主类和几个静态内部类，其中 RowData 这个类的属性，是会随着采集的表结构改变而改变的。所以后续会在这里通过反射来实现这个类的动态创建。

然后就是在 notifying 中使用 gson 完成解析，代码如下：

.notifying(record -> {
  Gson gson = new Gson();
  DebeziumEvent event = gson.fromJson(record.value(), DebeziumEvent.class);
  System.out.println(record);
  System.out.println("操作类型: " + event.op);
  System.out.println("数据库: " + event.source.db);
  System.out.println("表名: " + event.source.table);
  System.out.println("变更前 ID: " + (event.before != null ? event.before.id : "null"));
  System.out.println("变更后 ID: " + (event.after != null ? event.after.id : "null"));
}).build()