企业官网建设流程全解析

接业务方需求"把订单表实时同步到 ClickHouse 做报表"的那一刻，我下意识还是想沿用老方案：凌晨定时全量导出。但看到订单表已经 300 多万行、业务每天下班后还在操作数据时，我就知道这条路走不通了。后来我把目光投向 Flink CDC，用一条 SQL 把 MySQL binlog 的变更吃到 Flink，再实时写入 ClickHouse，插入、更新都能在秒级看到结果。这篇文章没有废话，从 binlog 原理一路讲到可复现的配置和代码，十分钟你就能自己搭出一条 MySQL 到 ClickHouse 的增量同步管道。

1. 为什么我淘汰了定时任务，改投 Flink CDC 阵营

先交代一下背景。业务库是 MySQL，跑订单、用户这类核心交易数据；分析侧用 ClickHouse，供报表、经营看板和大屏查询。最开始的同步方式很朴素：每天凌晨用定时任务把 MySQL 整表导出，再批量写进 ClickHouse。数据量小的时候还能忍，等订单表涨到两三百万行之后，问题开始集中爆发。

1.1 全量同步的三大痛点

第一是延迟。凌晨任务跑完，业务方早上看到的还是昨天 0 点的数据，一旦当天有运营活动，报表基本废掉。第二是资源竞争。全量导出要扫业务表、占连接、吃 IO，正好和生产高峰错不开，DBA 隔三差五来投诉。第三是同步脚本本身越来越难维护：新加一张表要改脚本，字段调整要改映射，删字段要谨慎小心，整个流程全靠人肉保证。

这些痛点叠加在一起，就变成了一个很现实的问题：能不能只同步变化的数据？能不能把延迟从 24 小时压到秒级？能不能不加任何业务代码？

1.2 替代方案对比和最终选择

我把市面上的方案拉了个表，逐个做了评估：

最终选 Flink CDC，理由很直接：它把最复杂的 binlog 解析封装好了，提供 Flink SQL 和 DataStream 两套 API，能复用 Flink 的 checkpoint、并行度、背压处理这些成熟能力。相比 canal 那套"解析—投递—消费—清洗—写入"的多跳链路，Flink CDC 把同步管道收敛成一个作业，逻辑简单，出问题也好定位。

2. 增量备份到底在备什么：binlog 与 Debezium 引擎

用 Flink CDC 之前，我建议你花两分钟搞清楚它背后的两个机制，否则配置参数的时候会一头雾水。这两个机制分别是 MySQL binlog 和 Debezium 解析引擎。

2.1 binlog 是增量数据唯一的可靠源头

MySQL 的 binlog（二进制日志）记录了所有数据变更操作。但记录方式有讲究：默认的 STATEMENT 格式只记录"执行了哪条 SQL"，这对 CDC 来说不够用；要拿到每行数据变更前后的具体值，必须把格式调成 ROW。

ROW 格式下，一次 UPDATE 会在 binlog 里留下两条事件：变更前镜像和变更后镜像；一次 DELETE 留下变更前镜像；INSERT 则只留下新值。Flink CDC 之所以能感知到"哪行被改成了什么"，靠的就是这些镜像。这里还有个容易被忽略的参数：binlog_row_image要设成FULL，否则部分列更新时，binlog 里可能只有被修改的列，其他列是空的，下游拿不到完整行。

同时还要注意 binlog 的保留时间。CDC 作业如果因为故障停了几个小时，重启后需要从上次的 binlog 位点继续消费；如果 binlog 在这期间被清理了，作业会直接启动失败。生产环境我习惯把保留时间调到至少 7 天，给故障恢复留足余量。

2.2 Flink CDC 内部藏着一个 Debezium

Flink CDC 的 MySQL connector 并不是从零写的解析器，它内置了 Debezium 这个成熟的 CDC 开源框架。整个采集过程可以理解成两个阶段：

第一阶段做全量快照。作业启动时，source 会先对目标表做一次一致性快照，把已有数据都读出来。Flink CDC 2.x 默认开启了增量快照机制，按主键分片并行读取，不需要对整张表加全局读锁，这个设计对生产库非常友好。

第二阶段吃增量日志。快照完成时，connector 会记下当前的 binlog 位点，之后所有变更都从这个位点开始消费。这个位点会保存到 Flink 的 checkpoint 里，作业任何一次重启，都能从最近一次的 checkpoint 恢复消费位置，这也是它能做到精确一次语义的基础。

打个比方：全量快照相当于把图书馆现有藏书全部抄一遍，同时记下"抄到哪本书时有人往书架塞了新书"，之后我只盯着新塞进来的动作，就能保持完全同步。

2.3 管道里流动的是 changelog，不是普通表数据

Flink CDC 从 binlog 解析出来的不是一张普通表，而是一条 changelog 流。流里的每条记录都带一个行类型标记：

• +I：插入的新行
• -U：更新前的旧行
• +U：更新后的新行
• -D：被删除的行

理解这个标记很重要，尤其是后面要把数据写进 ClickHouse 的时候。ClickHouse 是一套日志式存储，本身不擅长高频单行更新和删除，它对 changelog 的支持天然受限。所以建表方案需要绕一下，用 ReplacingMergeTree 加版本字段来模拟"新版本覆盖旧版本"的效果，这一点下一节会详细讲。

3. 动手前的环境准备：版本、MySQL 配置与 ClickHouse 建表

3.1 版本组合这样定最稳

Flink CDC 对版本组合比较敏感，我最初在版本上踩过不少软钉子。下面这组是我目前生产环境在用的，直接抄作业问题不大：

有一点要说明：Flink CDC 3.x 已经推出了新的架构，支持 YAML 配置方式，但 SQL 模式在 2.x 上最成熟，网上资料也最多，新手从 2.4.x 入手最不容易踩坑。

3.2 MySQL 侧三个必做动作

第一步，改配置并重启 MySQL：

[mysqld] server-id = 1 log-bin = mysql-bin binlog_format = ROW binlog_row_image = FULL binlog_expire_logs_seconds = 2592000

如果你用的是 MySQL 5.7，把最后一行换成expire_logs_days = 30。注意server-id不能和现有主从拓扑里的任何节点冲突，这里写 1 是因为我本地测试环境没有其他从节点，生产环境建议用 5400 以上的独立数字。

第二步，创建同步账号并授权：

CREATE USER 'flink_cdc'@'%' IDENTIFIED BY 'StrongPass_123'; GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'flink_cdc'@'%'; FLUSH PRIVILEGES;

SELECT用来做全量快照，REPLICATION SLAVE和REPLICATION CLIENT用来读 binlog，RELOAD和SHOW DATABASES是部分快照模式下的保险项。

第三步，确认 binlog 真的在写。执行SHOW VARIABLES LIKE 'binlog_format';看到 ROW 只是第一步，还要执行SHOW MASTER STATUS;两次，如果返回的 Position 在变大，说明 binlog 正在接收新事件。

3.3 ClickHouse 侧用 ReplacingMergeTree 接住 upsert

既然 ClickHouse 不擅长行级更新，我们的思路就变成：每次收到一条新版本数据，就往 ClickHouse 里追加一行，后台异步合并时保留版本号最大的那一行，用空间换语义。对应建表语句：

CREATE TABLE dwd.ods_users ( id UInt64, name String, age Int32, email String, updated_at DateTime64(3), cdc_version UInt64 ) ENGINE = ReplacingMergeTree(cdc_version) PRIMARY KEY (id) ORDER BY (id);

ReplacingMergeTree(cdc_version)的意思是：按照ORDER BY的 key（这里就是id）去重，合并时保留cdc_version最大的行。所以下游查询时通常要加FINAL关键字，或者通过物化视图进一步折叠。

关于版本字段的选取，我实践下来有三种做法，各有取舍：

4. 三步跑通一条 Flink SQL 同步管道

4.1 准备 Flink 运行时和依赖 jar

先下载 Flink 1.17.1 解压，本地起了 standalone 集群：

cd flink-1.17.1 ./bin/start-cluster.sh

把两个 jar 放进lib目录：flink-sql-connector-mysql-cdc-2.4.2.jar和clickhouse-jdbc-0.5.0.jar。因为我们是在启动集群之后才放的 jar，需要重启一次集群让 TaskManager 加载：

./bin/stop-cluster.sh ./bin/start-cluster.sh

重启后访问 8081 端口，能看到 Web UI 就说明集群起来了。

4.2 用 Flink SQL 注册 MySQL CDC 源表

进入 SQL Client：

./bin/sql-client.sh embedded

执行下面的建表语句：

CREATE TABLE mysql_users ( id BIGINT, name STRING, age INT, email STRING, updated_at TIMESTAMP(3) ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'localhost', 'port' = '3306', 'username' = 'flink_cdc', 'password' = 'StrongPass_123', 'database-list' = 'dwd', 'table-list' = 'dwd.users', 'server-time-zone' = 'Asia/Shanghai', 'scan.incremental.snapshot.enabled' = 'true', 'server-id' = '5401-5402', 'debezium.snapshot.mode' = 'initial' );

几个参数说明一下。scan.incremental.snapshot.enabled开启后，全量快照会按主键切片并行读，不锁整表；server-id给了一个范围，是给并行分片预留的 slot，不能和 MySQL 实例上已有的 slave id 冲突；debezium.snapshot.mode = initial表示先做完整快照，再无缝切换到 binlog 增量。

4.3 先用 print sink 验证 CDC 事件流

很多新手一上来就直接连 ClickHouse，出了问题反而分不清是 CDC 没采到数据，还是写入端有问题。我的习惯是先建一个 print sink，把 changelog 打到标准输出，花一分钟确认源头是通的：

CREATE TABLE print_sink ( id BIGINT, name STRING, age INT ) WITH ('connector' = 'print'); INSERT INTO print_sink SELECT id, name, age FROM mysql_users;

这时去 MySQL 里插入、修改、删除几条数据，然后在 Flink 的 TaskManager 日志或 SQL Client 结果里，应该能看到带+I、-U、+U、-D标记的记录。看到这些标记，说明 binlog 采集、解析、传输整条链路已经通了，后面的事只是"怎么把数据写进 ClickHouse"。

4.4 定义 ClickHouse 目标表并提交作业

验证完源头，就可以正式写 ClickHouse 了：

CREATE TABLE ch_users ( id BIGINT, name STRING, age INT, email STRING, updated_at TIMESTAMP(3), cdc_version BIGINT ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:clickhouse://localhost:8123/dwd', 'table-name' = 'ods_users', 'username' = 'default', 'password' = '', 'sink.buffer-flush.max-rows' = '500', 'sink.buffer-flush.interval' = '2s', 'sink.max-retries' = '3' );

然后提交同步作业：

INSERT INTO ch_users SELECT id, name, age, email, updated_at, CAST(UNIX_TIMESTAMP() * 1000 AS BIGINT) AS cdc_version FROM mysql_users;

这里有个非常重要的经验，直接决定了你能不能跑通：Flink 官方 JDBC connector 对 ClickHouse 默认是 append-only 语义，只认 INSERT，不认 CDC 的 update/delete changelog。如果你的 Flink 版本在提交时直接报类似The sink doesn't support consuming update changes的错误，别慌，这是预期行为。我在第 6 节专门整理了三种解法，第一次遇到可以直接跳到 6.1 看。

5. 实测验证：插入、更新、删除三条链路逐一核对

管道搭起来之后，我习惯用最笨也最可靠的办法验证：去 MySQL 改数据，然后看 ClickHouse 跟不跟。三个场景挨个测一遍。

5.1 插入：秒级到达 ClickHouse

先在 MySQL 里插一条：

INSERT INTO dwd.users (id, name, age, email, updated_at) VALUES (1001, '测试用户', 20, 'test@example.com', NOW());

等 2 到 3 秒（JDBC sink 默认攒批刷新，2 秒间隔），再去 ClickHouse 查：

SELECT * FROM dwd.ods_users WHERE id = 1001;

如果能查到这条数据，插入链路就通了。第一次看到秒级同步成功时，对比之前动辄半小时的全量任务，那种感受还是挺明显的。

5.2 更新：版本字段如何兜住 ClickHouse 不支持行更新

继续做更新：

UPDATE dwd.users SET age = 28 WHERE id = 1001;

这时候 ClickHouse 里大概率会出现两条id = 1001的记录，一条 age 是 20，一条 age 是 28。原因前面讲过：ClickHouse 不更新已有数据，只是追加新行，靠ReplacingMergeTree在后台异步合并。合并完成前，直接查明细表会看到"两条历史版本"。

所以查询时要么手动加FINAL：

SELECT * FROM dwd.ods_users FINAL WHERE id = 1001;

要么在测试环境手动触发一次合并：

OPTIMIZE TABLE dwd.ods_users FINAL;

生产环境我不会让人直接查这张明细表，通常会在它上面再建物化视图做折叠，保证报表查询自动拿到最新版本。

5.3 删除：一个需要提前想清楚的业务取舍

最后测删除：

DELETE FROM dwd.users WHERE id = 1001;

这一步在不同 sink 方案下表现完全不同。如果你用了第 6.1 节推荐的 DataStream 过滤方案，删除事件会被过滤掉，ClickHouse 里这条数据依然存在。对分析型报表来说，这通常可以接受：订单一旦产生，即使业务上退单，也只是状态字段变成"已取消"，物理删除本来就罕见。

如果你的业务确实需要物理删除同步，那就得在 ClickHouse 侧引入删除标记字段，或者定期用外部任务清理孤儿数据。这里没有银弹，关键是上线前要和业务方说清楚"删除到底同不同步"，别等上线后才发现两边数据对不上。

6. 生产化路上绕不开的五个坑

跑通 demo 只是第一步，真正上线还会遇到下面这些坑。我把它们按出现频率排了个序。

6.1 官方 JDBC sink 不认 update 事件：三种解决方案

这个问题我在第 4.4 节预告过，核心是：MySQL CDC 吐出来的是 changelog，官方 JDBC sink 对 ClickHouse 默认只支持追加写，二者语义对不上，作业会直接报错。

我的三种处理方式，按推荐程度从低到高排列：

第一种，给 ClickHouse sink 表加上主键PRIMARY KEY (id) NOT ENFORCED，让 JDBC sink 进入 upsert 模式。这个方案在 MySQL、PostgreSQL 上很顺，但 ClickHouse 的 JDBC 方言对标准 upsert SQL 支持不完整，不同驱动版本表现还不一样，生产环境我不敢赌。

第二种，把 ClickHouse 当成纯追加存储，用printsink 验证 CDC 数据流，然后写一个专门的同步程序消费。链路是通的，但等于绕开了 Flink 的强项，多了一套组件。

第三种，也是我在生产用的方案：保留 Flink SQL 的源表和计算逻辑，但把写入端换成轻量的 DataStream Sink，过滤掉更新前镜像和删除事件，只把+I/+U的新镜像 INSERT 进 ClickHouse。核心代码大概这样：

DataStreamSource<String> cdcStream = env.addSource( MySqlSource.<String>builder() .hostname("localhost") .port(3306) .databaseList("dwd") .tableList("dwd.users") .username("flink_cdc") .password("StrongPass_123") .deserializer(new JsonDebeziumDeserializationSchema()) .build()); cdcStream .filter(json -> { // op 字段：c 为 insert，u 为 update，d 为 delete return !"d".equals(JsonNode.readTree(json).get("op").asText()); }) .addSink(new ClickHouseBatchSink());

ClickHouseBatchSink内部逻辑很简单：解析 JSON 里的after字段，拼成 INSERT 语句，攒够 500 条或 2 秒批量执行一次。配合 ClickHouse 表上的版本字段，ReplacingMergeTree 会自动保留最新版本，语义和"业务视角的最终一致"完全对得上。

6.2 DDL 变更导致作业失败的处理流程

MySQL 源表加一个字段，Flink CDC 作业经常直接失败。这是因为 CDC 在启动时已经把源表 schema 缓存下来了，源端表结构一变，解析出来的数据对不上 sink 定义，作业就会因为反序列化异常退出。

我现在的处理流程是：先改 ClickHouse 目标表，再加 Flink SQL sink 表定义，最后重启作业。顺序不能反，否则即使作业不挂，数据也会写错列。如果你是 Flink CDC 2.4 以上版本，可以关注下它逐步增强的 schema evolution 能力，但现阶段我依然建议把 DDL 变更当成一次有计划的变更管理来做，别指望全自动。

6.3 类型映射、时区和 NULL 值

类型映射是第二高频的坑。MySQL 的decimal(10,2)到 ClickHouse 要手动确认是Decimal(18,2)还是Decimal(38,8)，精度没对齐，同步不会报错，但报表数字会错，而且特别难排查。时间类型同理，MySQL 的datetime精度和 ClickHouse 的DateTime64(3)要保持一致，否则查询端出来的时间差着量级。

时区问题也值得单独说。MySQL 连接参数里我显式配了server-time-zone = 'Asia/Shanghai'，因为默认时区解析一旦偏掉，所有时间字段都会整体偏移，看起来毫无规律，实际全是时区问题。NULL 值则要注意 Debezium 默认对 NULL 的处理，某些版本会把 NULL 包装成__debezium_unavailable_value，下游不认识就会反序列化失败。

6.4 Checkpoint 没开，重启一次丢一段数据

Flink CDC 的位点恢复完全依赖 checkpoint。如果作业配置里没开 checkpoint，作业一重启，source 会从最早的位点或者最新位点重新拉，要么丢数据，要么重复大量数据。

在 SQL Client 里提交作业前加上这几行：

SET 'execution.checkpointing.interval' = '30s'; SET 'execution.checkpointing.mode' = 'EXACTLY_ONCE'; SET 'state.backend.type' = 'rocksdb';

RocksDB 需要把flink-statebackend-rocksdb-1.17.1.jar也放进 Flink 的lib目录。状态量一大，RocksDB 的优势就体现出来了，内存占用可控，也不容易出现 OOM。

6.5 作业正常但数据不动：一步步定位的排查链路

最让人头疼的场景不是作业报错，而是作业显示 RUNNING、checkpoint 也正常，但 ClickHouse 里就是没数据。我总结了一套排查顺序，按这个顺序走基本十分钟内能定位：

1. 先看 MySQL binlog 有没有新事件。执行SHOW MASTER STATUS;，多执行几次，Position 不变说明源端压根没有新写入，不是 Flink 的问题。
2. 看 Flink Web UI 里 source 节点的currentFetchEventTimeLag指标。这个值持续增大，说明 binlog 已经读到但下游消费慢，属于背压问题；值很小但目标表没数据，问题大概率在 sink。
3. 把作业临时切换到printsink 跑一遍。print 能看到最新的事件流，如果 print 有数据而 ClickHouse 没有，那就是 sink 端的问题，重点检查 JDBC url、表名、攒批参数。
4. 手动跑一次不带条件的SELECT * FROM dwd.ods_users FINAL，排除 ReplacingMergeTree 还没合并导致"看起来没变化"的假象。
5. 最后确认 ClickHouse 表的版本字段是不是一直在增长。如果cdc_version都一样，说明版本来源选得不对，同一主键的多条变更会互相覆盖，这也是最隐蔽的坑。

这套链路我至少用过十次，每次都能快速缩小范围。核心原则是：划分边界、逐段验证，别一开始就在某个环节里死磕。

最后分享一个我自己的习惯：任何 CDC 管道上线前，我都会在测试库上完整跑一遍"插入、更新、删除"三条用例，并且在 ClickHouse 侧每天拉一次主键唯一性校验。ReplacingMergeTree 是异步合并的，FINAL查询和后台合并存在延迟，报表查询尽量走物化视图，不要直接查明细表。这套方案我跑了半年多，最省心的就是它不需要业务表加任何字段，binlog 里有什么就同步什么；最大的教训则是"先验证再上线"，尤其是删除语义这种业务陷阱，想清楚再动手。