4、Flink CDC核心API详解：DataStream API与Table/SQL API对比、SourceBuilder构建、DeserializationSchema自定义反序列化

这一章，我们来啃Flink CDC最核心的API。说实话，很多初学者一上来就被各种API搞晕了——到底用DataStream还是Table/SQL？SourceBuilder怎么配？反序列化又是什么鬼？

别急，我一个一个给你拆开讲。先看一张总览图，把今天要聊的三个知识点串起来。

4.1 DataStream API vs Table/SQL API：怎么选？

我刚开始用Flink CDC的时候，也纠结过这个问题。后来做多了发现，这俩没有的优劣，关键看场景。

4.2 SourceBuilder：构建CDC Source的正确姿势

不管用哪种API，底层都离不开SourceBuilder。我记得第一次用的时候，参数配错了，结果全量阶段跑完增量阶段一直没数据——排查了半天才发现是startupOptions设成了initial但没给Server ID。

来看一个标准的构建方式：

// 构建 MySQL CDC Source
MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
    .hostname("localhost")
    .port(3306)
    .databaseList("mydb")                 // 监控的数据库
    .tableList("mydb.orders")             // 监控的表，格式 db.table
    .username("cdc_user")
    .password("cdc_pass")
    .serverTimeZone("Asia/Shanghai")
    .startupOptions(StartupOptions.initial())
    .deserializer(new JsonDebeziumDeserializationSchema())
    .build();

这里有几个关键参数，我重点说一下：

• hostname/port：MySQL地址，注意网络连通性，别配成localhost了。
• databaseList/tableList：支持正则，比如"mydb\\..*"匹配mydb下所有表。
• startupOptions：有initial（先全量后增量）、earliest（从最早binlog开始）、latest（从当前binlog开始）、timestamp（指定时间戳）。
• deserializer：反序列化器，后面重点讲。

4.3 DeserializationSchema：自定义反序列化

默认的JsonDebeziumDeserializationSchema会把CDC事件转成JSON字符串。但很多时候，我们需要自定义格式——比如只取变更后的数据、或者把Debezium的复杂结构拍平。

为什么要自定义？我给你举个例子。默认的CDC JSON长这样：

{
  "before": null,
  "after": {
    "id": 1001,
    "name": "张三",
    "amount": 500.00
  },
  "source": {
    "db": "mydb",
    "table": "orders",
    "ts_ms": 1700000000000
  },
  "op": "c"
}

这个结构太冗余了。如果我只关心after字段和操作类型，完全可以自己写一个反序列化器，直接输出一个POJO。

来看代码：

public class OrderDeserializationSchema
    implements DeserializationSchema<OrderEvent> {

    @Override
    public void open(DeserializationSchema.InitializationContext context) {
        // 可以在这里初始化资源，比如连接池
    }

    @Override
    public OrderEvent deserialize(byte[] message) throws IOException {
        // 1. 解析Debezium的JSON
        JsonNode root = JsonParser.parseBytes(message);
        JsonNode after = root.get("after");
        String op = root.get("op").asText();

        // 2. 只处理插入和更新
        if ("c".equals(op) || "u".equals(op)) {
            OrderEvent event = new OrderEvent();
            event.setId(after.get("id").asLong());
            event.setName(after.get("name").asText());
            event.setAmount(after.get("amount").asDouble());
            event.setOpType(op);
            return event;
        }
        // 删除事件可以忽略，或者特殊处理
        return null;
    }

    @Override
    public boolean isEndOfStream(OrderEvent nextElement) {
        // CDC流是无界的，永远返回false
        return false;
    }

    @Override
    public TypeInformation<OrderEvent> getProducedType() {
        return TypeInformation.of(OrderEvent.class);
    }
}

然后这样用：

MySqlSource<OrderEvent> source = MySqlSource.<OrderEvent>builder()
    .hostname("localhost")
    .port(3306)
    .databaseList("mydb")
    .tableList("mydb.orders")
    .username("cdc_user")
    .password("cdc_pass")
    .deserializer(new OrderDeserializationSchema())
    .build();

DataStream<OrderEvent> stream = env.fromSource(source, 
    WatermarkStrategy.noWatermarks(), "MySQL CDC Source");

你看，这样下游拿到的就是干净的OrderEvent对象，直接做业务处理，不用再解析JSON了。

4.4 两种API的CDC实战对比

说了这么多，咱们直接看代码对比。同样是读取MySQL的orders表，两种写法差异很大。

Table/SQL API 方式：

// 创建CDC表
tableEnv.executeSql("CREATE TABLE orders_cdc (" +
    "  id BIGINT," +
    "  name STRING," +
    "  amount DECIMAL(10, 2)," +
    "  PRIMARY KEY (id) NOT ENFORCED" +
    ") WITH (" +
    "  'connector' = 'mysql-cdc'," +
    "  'hostname' = 'localhost'," +
    "  'port' = '3306'," +
    "  'database-name' = 'mydb'," +
    "  'table-name' = 'orders'," +
    "  'username' = 'cdc_user'," +
    "  'password' = 'cdc_pass'" +
    ")");

// 直接查
Table result = tableEnv.sqlQuery("SELECT id, name, amount FROM orders_cdc");
tableEnv.toDataStream(result, OrderEvent.class).print();

DataStream API 方式：

MySqlSource<OrderEvent> source = MySqlSource.<OrderEvent>builder()
    .hostname("localhost")
    .port(3306)
    .databaseList("mydb")
    .tableList("mydb.orders")
    .username("cdc_user")
    .password("cdc_pass")
    .deserializer(new OrderDeserializationSchema())
    .build();

DataStream<OrderEvent> stream = env.fromSource(source, 
    WatermarkStrategy.noWatermarks(), "MySQL CDC Source");

stream
    .filter(event -> event.getAmount() > 100)
    .map(event -> new EnrichedOrder(event, getRegion(event.getId())))
    .print();

看到区别了吗？SQL方式简洁，但做复杂过滤和关联就比较吃力。DataStream方式代码多几行，但你可以随意加filter、map、keyBy，甚至用ProcessFunction做状态管理。