Flink作业无法实时捕获DB新增数据的解决方案咨询

你的核心问题是当前JDBC Source仅能一次性全量读取数据，无法实时捕获数据库的新增/更新，且不想通过轮询增加数据库负载。最优方案是使用Flink CDC（Change Data Capture），它基于数据库的变更日志（如MySQL Binlog、PostgreSQL WAL）实现无轮询的实时数据捕获，完全匹配你的需求。

为什么选Flink CDC？ #

• 无需定时轮询，直接监听数据库的变更日志，几乎无额外数据库负载
• 实时捕获INSERT/UPDATE/DELETE所有变更操作
• 支持主流数据库：MySQL、PostgreSQL、Oracle、SQL Server等
• 官方原生支持，稳定性和兼容性有保障

具体实现步骤（以MySQL为例） #

1. 添加依赖 #

在你的项目中引入Flink CDC Connector的Maven依赖：

<dependency>
    <groupId>com.ververica</groupId>
    <artifactId>flink-connector-mysql-cdc</artifactId>
    <version>2.4.2</version> <!-- 请使用与Flink版本匹配的最新版本 -->
</dependency>

2. 替换原有JDBC Source为CDC Source #

将作业2中的FooJdbcSource替换为MySQL CDC Source，示例代码如下：

import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;

// 构建MySQL CDC Source
MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
        .hostname("你的数据库地址")
        .port(3306)
        .databaseList("你的数据库名") // 指定要监控的数据库
        .tableList("你的数据库名.foo_table") // 指定要监控的表
        .username("数据库用户名")
        .password("数据库密码")
        .deserializer(new JsonDebeziumDeserializationSchema()) // 将变更事件序列化为JSON
        .build();

// 从CDC Source读取数据
DataStream<String> stream = env.fromSource(
        mySqlSource,
        WatermarkStrategy.noWatermarks(),
        "mysql-cdc-source"
);

// 处理变更数据并发送到Topic2
stream.map(jsonStr -> {
    // 解析JSON格式的变更事件，构建AxonMessage
    // 示例：根据Debezium的JSON结构提取新增/更新的数据
    // 这里需要根据你的业务逻辑调整解析逻辑
    FooModel foo = JSON.parseObject(jsonStr, FooModel.class);
    AxonMessage message = new AxonMessage(foo);
    message.setTopic(Constants.PRODUCER_TOPIC_NAME);
    message.setKey(FooModel.class);
    return message;
}).sinkTo(axon.sink())
        .name("foo-kafka-sink")
        .uid("foo-kafka-sink");

3. 数据库配置要求 #

确保你的MySQL数据库开启了Binlog：

• 在my.cnf中配置：

  log-bin=mysql-bin # 开启Binlog
  binlog-format=ROW # 必须使用ROW格式，才能捕获行级变更
  server-id=1 # 唯一ID，不能和其他数据库实例重复
  

替代方案（如果数据库不支持CDC） #

如果你的数据库不支持变更日志（如部分老旧数据库或小众数据库），可以考虑：

• 数据库触发器+消息队列：在数据库表上创建触发器，当数据变更时将事件写入消息队列（如Kafka），再用Flink消费该队列。但这种方案会增加数据库的触发器负载，且可靠性不如CDC，仅作为备选。

总结 #

Flink CDC是解决你问题的最佳方案，它完全避免了轮询带来的数据库负载，同时实现了实时的变更捕获，完美匹配你的业务需求。

内容的提问来源于stack exchange，提问作者Khushboo