4. Flink CDC与数据源接入：Flink CDC原理、MySQL/Kafka CDC实战、Debezium集成、增量数据捕获的最佳实践

数据网格的核心，是把数据当成产品来管理。那数据从哪来？

说白了，得从业务系统里实时捞出来。我见过太多团队，上来就搞批处理，每天凌晨跑一次ETL。结果业务方早上问“昨天的订单数据怎么还没更新？”——这种场景，你经历过就懂了。

CDC（Change Data Capture，变更数据捕获）就是来解决这个问题的。它让你能实时感知数据库里的每一行变化，然后把这些变化流进数据网格里。今天我们就聊聊Flink CDC，以及它背后的那些事。

4.1 Flink CDC 的核心原理

Flink CDC 不是什么新概念。它本质上是一个 Source Connector，能读取数据库的 binlog（比如 MySQL 的 binlog），然后把 INSERT、UPDATE、DELETE 事件转换成 Flink 的 DataStream。

它的底层依赖 Debezium。Debezium 是一个 CDC 框架，专门干这件事。Flink CDC 把它封装成了 Flink 的 Source 接口，让你能在 SQL 或 DataStream API 里直接用。

嗯，这里要注意。快照阶段和增量阶段之间，有一个“无缝切换”的问题。我早期用 Flink CDC 1.x 的时候，就遇到过快照读完、增量没接上的情况，丢了几秒的数据。后来社区优化了，现在 2.x 版本已经解决了这个问题。

4.2 MySQL CDC 实战

先来个最简单的例子。假设你有一个 MySQL 表 orders，你想把它的变更实时同步到 Kafka 里。

第一步：开启 MySQL 的 binlog

MySQL 默认可能没开 binlog。你得在 my.cnf 里配一下：

[mysqld]
server-id = 1
log_bin = mysql-bin
binlog_format = ROW
binlog_row_image = FULL
expire_logs_days = 7

注意，binlog_format 必须设为 ROW。如果是 STATEMENT 或 MIXED，Flink CDC 读不到完整的数据变化。

第二步：创建 Flink CDC 表

在 Flink SQL 里，你可以这样定义：

CREATE TABLE orders_cdc (
  order_id BIGINT,
  user_id BIGINT,
  product_id BIGINT,
  amount DECIMAL(10, 2),
  status STRING,
  create_time TIMESTAMP(3),
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'localhost',
  'port' = '3306',
  'username' = 'cdc_user',
  'password' = 'cdc_pass',
  'database-name' = 'shop',
  'table-name' = 'orders',
  'server-id' = '5400-5404'
);

第三步：写入 Kafka

再定义一个 Kafka 表，然后一条 INSERT INTO 搞定：

CREATE TABLE orders_kafka (
  order_id BIGINT,
  user_id BIGINT,
  product_id BIGINT,
  amount DECIMAL(10, 2),
  status STRING,
  create_time TIMESTAMP(3)
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders_cdc',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'debezium-json'
);

INSERT INTO orders_kafka SELECT * FROM orders_cdc;

你看，就这么几行 SQL，MySQL 到 Kafka 的实时同步就搭好了。

4.3 Kafka CDC 与 Debezium 集成

有些场景下，你的数据源不是 MySQL，而是 Kafka。比如，你已经用 Debezium 把 MySQL 的变更流到了 Kafka，现在想用 Flink 消费这个流。

Debezium 输出的消息格式是它自己的 JSON 结构。里面包含了 before、after、op 等字段。Flink 的 debezium-json 格式能自动解析它。

举个例子，Debezium 输出的消息长这样：

{
  "before": null,
  "after": {
    "order_id": 1001,
    "user_id": 42,
    "amount": 199.00,
    "status": "PAID"
  },
  "op": "c",
  "ts_ms": 1680000000000
}

在 Flink 里，你可以直接用 debezium-json 格式来解析：

CREATE TABLE orders_debezium (
  order_id BIGINT,
  user_id BIGINT,
  amount DECIMAL(10, 2),
  status STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'dbserver1.shop.orders',
  'properties.bootstrap.servers' = 'localhost:9092',
  'value.format' = 'debezium-json'
);

4.4 增量数据捕获的最佳实践

做 CDC 项目，坑不少。我踩过一些，分享给你。

1. 全量快照的性能问题

Flink CDC 启动时，会先做一次全量快照。如果表很大（比如几亿行），这个快照可能会跑很久，而且会占用数据库连接。

我的建议是：

• 用 scan.incremental.snapshot.enabled 参数，开启增量快照模式。它会分片读取，不会锁表。
• 如果表实在太大，考虑先做一次离线全量同步，再启动 CDC 只做增量。

2. 主键的重要性

Flink CDC 依赖主键来保证 exactly-once 语义。没有主键的表，更新和删除操作可能会出问题。

我曾经接手过一个项目，订单表没有主键，结果 CDC 同步后，数据对不上。后来加了主键才解决。

3. 数据类型映射

MySQL 的 DECIMAL、DATETIME、JSON 等类型，映射到 Flink 时要注意。比如 DATETIME 默认是 TIMESTAMP(3)，但 MySQL 的精度可能是 6 位。你最好在 DDL 里显式指定精度。

4. 监控与告警

CDC 链路一旦断了，数据延迟会越来越大。我建议你监控这几个指标：

指标	说明	告警阈值
current-fetch-event-time-lag	当前事件与最新 binlog 的时间差	> 30 秒
snapshot-phase	是否处于快照阶段	超过预期时间
binlog-position	当前读取的 binlog 位置	长时间不变

4.5 整体架构图

下面这张图，展示了 Flink CDC 在数据网格中的典型位置。它从业务数据库捕获变更，经过 Flink 处理后，流入数据网格的各个数据产品中。

4.6 避坑指南

最后，总结几个我踩过的坑，你遇到了可以少走弯路。

• 不要用 root 用户连 CDC： 给 CDC 单独建一个用户，只给 SELECT、REPLICATION SLAVE、REPLICATION CLIENT 权限。安全第一。
• 小心 DDL 变更： 如果源表加了列，Flink CDC 不会自动感知。你需要重启作业，或者用 scan.incremental.snapshot.chunk.key-column 参数处理。
• Kafka 消息大小限制： 如果一行数据很大（比如包含大字段），Kafka 默认的 1MB 消息大小可能不够。调大 max.message.bytes 和 replica.fetch.max.bytes。
• 时区问题： MySQL 的 TIMESTAMP 存的是 UTC，但 Flink 默认用本地时区。你最好在 Flink 配置里统一设置时区，比如 table.local-time-zone = Asia/Shanghai。

好了，CDC 这块就聊到这。你只要把原理搞懂，把 Debezium 和 Flink CDC 的配置摸熟，数据源接入这块基本就稳了。数据网格的“实时性”，就靠它撑着了。