企业官网建设流程全解析

做MySQL到ClickHouse的数据同步，早期我的方案特别土：每天凌晨定时任务全量抽取，业务低峰再扫一遍增量字段。数据量小的时候看不出问题，一旦单表数据过千万，全量扫描对源库的压力肉眼可见。更难受的是增量依赖业务字段（比如update_time）去判断，漏数据、重复数据都是家常便饭。直到我把Flink CDC引入链路，用binlog做实时增量捕获，配合ClickHouse做分析侧存储，才算真正把这件事理顺。这篇文章直接把我的落地过程、配置细节和踩坑经历全部写出来，十分钟读完，照着抄就能用。

1. 项目概述与核心需求解析

1.1 这个需求到底在解决什么问题

先说清楚需求本质。MySQL是OLTP系统，擅长事务处理，订单、用户这些业务数据都跑在它上面。ClickHouse是OLAP系统，擅长海量数据的聚合分析。业务跑久了，分析侧需要一份和线上库保持一致的副本，供报表、BI、实时大屏查询。我要做的不是每天导一次全量，而是让MySQL里的每一次INSERT、UPDATE、DELETE都尽可能实时地同步到ClickHouse，同时保证两边数据最终一致。

这个需求里有三个关键点容易被忽略。第一，不能给源库加太大负担，尤其是千万级以上的大表，全表扫描一次代价极高，还会拖慢线上业务。第二，增量不能靠业务字段硬猜，否则删数据、改主键、跨时区更新时间这些场景都会让数据对不上账。第三，同步过程要容忍网络抖动和任务重启，不能因为一次抖动就丢数据或者重复灌数据。这三个点，恰恰是Flink CDC这套方案能同时覆盖的。

1.2 为什么Flink CDC是当前性价比最高的方案

我当时其实是把市面上的方案都过了一遍才做的选择。

第一个是定时批处理，用脚本每天全量导入一次ClickHouse。数据量在百万级以内可以忍，千万级之后源库扛不住，而且分析数据的实时性基本为零，凌晨跑完的报表白天才能看，业务要个实时数据只能干瞪眼。

第二个是Canal监听binlog，自己写消费端同步到ClickHouse。Canal本身很成熟，但问题在于需要自己维护一套消费程序，断点续传、幂等写入、类型转换、异常重试这些全都要自己写。做出来不难，做好很难，光是一个"任务挂了怎么恢复不丢数据"就能折腾半个月。

第三个是DataX这类离线同步工具，适合一次性批量迁移，做实时增量不合适，它的设计目标就不是干这个的。

Flink CDC的优势在于，它内部基于binlog解析（底层依赖Debezium），把全量快照和增量日志无缝衔接，同时Flink框架本身提供Exactly-Once语义和Checkpoint机制。配合Flink SQL，我只需要声明一张源表和一张目标表，一句INSERT INTO就能启动同步任务。这个开发效率，是自研消费端完全没法比的。唯一需要付出的学习成本就是Flink的基本概念，但只要上手一次，后面做任何数据同步需求都会快很多。

提示：如果你的源库不是MySQL，而是PostgreSQL或Oracle，Flink CDC同样支持，只是连接器参数不同。这篇文章后续内容以MySQL为主线展开，其他数据库的原理是相通的。

2. 环境准备与架构设计

2.1 版本选型与兼容性坑

版本是第一个坑。Flink CDC连接器的版本和Flink主版本必须匹配，否则任务能提交，但运行时会各种奇怪报错。我这次用的是Flink 1.18配合flink-cdc-connector 2.4.2，这套组合在社区里用的人最多，文档全，遇到问题也容易搜到答案。如果你用Flink 1.17，记得选对应的连接器版本，不要无脑下最新版，最新版往往要求更高的Flink版本。

ClickHouse这边我用的21.8 LTS版本，这个版本的数据类型和SQL语法对Flink写入兼容性更好。MySQL要注意，如果生产环境还是5.7，功能上没问题；如果是8.0，要注意认证插件的问题。MySQL 8.0默认使用caching_sha2_password认证插件，Flink CDC连接器对老版本MySQL认证方式兼容更稳定，所以建议在MySQL里单独建一个使用mysql_native_password插件的同步账号，避免认证握手失败。

Flink本身需要JDK 11以上，我用JDK 11跑Flink 1.18很稳定。如果公司环境强制JDK 8，那建议把Flink版本降到1.14左右，配套的CDC连接器版本也要相应调整。这里面的兼容矩阵，官方文档有版本对应表，动手前务必先查清楚，不然会在环境问题上浪费很多时间。

2.2 整体架构与数据链路拆解

整个链路是标准的四层结构。

第一层是MySQL源库，开启binlog且格式为ROW。Flink CDC连接器启动后会先做一次全量快照，然后自动切换到binlog消费增量，这个切换过程对应用层是透明的。

第二层是Flink任务，通过flink-cdc-connector读取binlog事件，在内部完成反序列化、类型转换，然后交给下游算子。任务的并发度、Checkpoint间隔、状态后端，决定了同步的吞吐和容错能力。

第三层是ClickHouse目标集群，这里有个关键设计：MySQL的主键更新在ClickHouse里要用ReplacingMergeTree表引擎配合版本字段处理，否则UPDATE只会变成追加一行，重复数据会直接污染分析结果。

第四层是监控与运维，Flink的Web UI可以看到每个任务的延迟指标，ClickHouse的system.query_log可以排查写入性能。

这个架构最值钱的地方在于，全量快照和增量同步在一个任务里完成，不需要先手动导全量再启动增量。Flink CDC在启动时，如果MySQL表里已经有数据，它会先快速扫一遍全量，同时记录binlog位点，全量完成后无缝衔接增量，整个过程不会丢数据也不会重复。

3. MySQL侧准备工作

3.1 binlog参数配置与账号授权

MySQL侧是整个链路的地基，地基不稳，上面全白搭。

先改my.cnf，核心参数就四个：

[mysqld] server_id = 123456 log_bin = mysql-bin binlog_format = ROW binlog_row_image = FULL

server_id必须设置为在整个集群内唯一的数字，这是个特别容易踩的坑。如果多台同步机器用了相同的server_id，MySQL会误判为同一个复制源，binlog拉取直接报错。所以每一条同步链路都要分配一个独立ID。

binlog_format必须改成ROW，因为Statement格式记录的是SQL语句，不能精确还原每行数据的变更，在UPDATE和DELETE场景下会丢字段信息。binlog_row_image要设成FULL，确保binlog里包含变更行的所有列值，而不是只包含主键和变更列。

expire_logs_days建议设置成7天左右。太短的话，Flink任务故障时间一长，binlog位点已经过期，任务恢复时扫描不到历史日志，只能重新全量；太长则浪费磁盘空间，这个参数可以根据你的故障恢复SLA来定。

改完配置重启MySQL后，验证一下配置生效：

SHOW VARIABLES LIKE 'binlog_format'; SHOW VARIABLES LIKE 'binlog_row_image';

接着创建同步账号：

CREATE USER 'cdc_user'@'%' IDENTIFIED WITH mysql_native_password BY '你的密码'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'cdc_user'@'%'; FLUSH PRIVILEGES;

这里要特别注意，SELECT权限是Flink CDC做全量快照时读数据用的，REPLICATION SLAVE和REPLICATION CLIENT权限是拉取binlog用的。缺一个权限，任务就会在启动阶段直接报Access denied。MySQL 8.0里REPLICATION SLAVE这个旧名称仍然可用，如果在MariaDB上，权限名称又不一样，需要单独确认。

3.2 配置生效验证

配置改完，怎么确认真的生效？先在MySQL命令行里执行SHOW MASTER STATUS，能看到当前binlog文件名和位点，说明binlog正常工作。然后执行一条UPDATE语句，再用binlog解析工具查看日志内容，确认记录的是ROW格式而不是Statement格式。

判断server_id有没有冲突有个笨办法：同时启动两个测试连接去拉binlog，如果第二个连接报错并提示server_id重复，就说明冲突了。这个问题在生产环境特别容易半夜爆发，一定要在测试阶段提前验证。

4. Flink CDC同步任务实现

4.1 源表与目标表的Schema映射

到了真正动手写任务的时候。第一步是把MySQL的表结构翻译成ClickHouse建表语句，这个映射关系不能想当然，类型对应错了数据会直接灌不进去。

用一个典型订单表举例，MySQL侧建表：

CREATE TABLE `orders` ( `id` BIGINT NOT NULL, `order_no` VARCHAR(64) NOT NULL, `user_id` INT NOT NULL, `amount` DECIMAL(10,2) NOT NULL, `status` TINYINT NOT NULL, `remark` TEXT, `created_at` DATETIME NOT NULL, `updated_at` DATETIME NOT NULL, PRIMARY KEY (`id`) );

对应的ClickHouse建表：

CREATE TABLE default.orders ( `id` Int64, `order_no` String, `user_id` Int32, `amount` Decimal(10, 2), `status` Int8, `remark` Nullable(String), `created_at` DateTime, `updated_at` DateTime, `_version` UInt64 ) ENGINE = ReplacingMergeTree(_version) ORDER BY id;

几个映射要点单独说一下。MySQL的BIGINT对应ClickHouse的Int64，INT对应Int32，TINYINT对应Int8，VARCHAR和TEXT都对应String，DECIMAL直接对应Decimal且小数位要一致。DATETIME对应DateTime，DATE对应Date。

ClickHouse的字段默认是非空，但MySQL的TEXT字段可能是NULL，所以映射到ClickHouse时，源表允许为空的字段都要用Nullable包一层，否则写入时遇到NULL会直接失败。这是个非常隐蔽的坑，我见过有人在这里调试到半夜。

还有一点，处理UPDATE场景必须用ReplacingMergeTree引擎并加版本字段。我在表结构里加了一个_version字段，用来承载binlog里的事件时间戳，ClickHouse会在后台根据version去重，保留最新版本的那一行。不用这个引擎的话，MySQL里的UPDATE到了ClickHouse就是追加新行，金额、状态这些字段会同时存在多份，报表一查就全乱了。

4.2 Flink SQL建表与提交任务

表结构准备好之后，就是写Flink SQL了。我用Flink SQL Client方式提交，适合快速开发和验证。先创建源表：

CREATE TABLE orders_source ( id BIGINT, order_no STRING, user_id INT, amount DECIMAL(10, 2), status TINYINT, remark STRING, created_at TIMESTAMP, updated_at TIMESTAMP, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = '127.0.0.1', 'port' = '3306', 'username' = 'cdc_user', 'password' = '你的密码', 'database-name' = 'app_db', 'table-name' = 'orders', 'scan.startup.mode' = 'initial' );

注意scan.startup.mode这个参数，它有三个常用值：initial表示全量加增量，任务启动时先扫全量再切增量；earliest-offset表示直接从最早的binlog开始消费；latest-offset表示只消费启动之后的新变更。第一次跑建议用initial，把历史数据和增量一次性对齐，后面重建任务时才考虑latest-offset。

再创建ClickHouse目标表：

CREATE TABLE orders_sink ( id BIGINT, order_no STRING, user_id INT, amount DECIMAL(10, 2), status TINYINT, remark STRING, created_at TIMESTAMP, updated_at TIMESTAMP, `_version` UInt64 ) WITH ( 'connector' = 'clickhouse', 'url' = 'clickhouse://127.0.0.1:8123', 'database-name' = 'default', 'table-name' = 'orders', 'sink.batch-size' = '1000', 'sink.flush-interval' = '1000' );

最后一句SQL启动同步：

INSERT INTO orders_sink SELECT id, order_no, user_id, amount, status, remark, created_at, updated_at, now() * 1000 FROM orders_source;

注意最后我用了now() * 1000来生成版本号，这个版本号写进ClickHouse的_version字段，用于ReplacingMergeTree去重。生产环境更严谨的做法是用binlog自带的变更时间戳，Flink CDC内部能拿到这个时间，但不同版本的连接器暴露方式不太一样，为了演示简单我用now()替代，生产上建议替换成事件时间。

任务提交之后，Flink会先执行全量快照，源表和目标表的数据量会快速对齐。全量完成后进入增量阶段，MySQL里的每一条变更，一两秒内就会出现在ClickHouse里。如果不想用SQL Client，也可以改用DataStream API，灵活度更高，但开发量会上去。我的建议是：简单的表同步用Flink SQL，涉及复杂清洗逻辑再用DataStream，没必要一上来就上API。

4.3 Checkpoint与一致性问题

同步任务跑起来容易，但要做到"挂了不丢数据、重启不重复数据"，就必须认真配置Checkpoint。

Flink SQL Client里可以直接设置：

SET 'execution.checkpointing.interval' = '10s'; SET 'execution.checkpointing.mode' = 'EXACTLY_ONCE'; SET 'state.backend' = 'rocksdb'; SET 'state.checkpoints.dir' = 'file:///data/flink/checkpoints';

Checkpoint间隔不建议设太短，太短会导致状态频繁持久化，拖慢整体吞吐；太长则故障恢复时丢失的数据窗口大。10秒到30秒是我试下来的舒适区间。

state.backend用RocksDB，特别是源表数量多、状态大时，基于堆的HashMap状态后端容易导致TaskManager频繁Full GC甚至OOM。RocksDB把状态落到本地磁盘，内存占用可控，生产环境几乎都是它。

还有一个关键认知：ClickHouse的写入本身不支持事务，所以Exactly-Once到ClickHouse是靠幂等写入实现的，也就是"写重复了也没关系，最终能被ReplacingMergeTree去重"。这也是为什么前面强调一定要有_version字段并且版本号单调递增。没有这个去重兜底，Checkpoint恢复后重复写入的数据会直接翻倍，报表数据错乱是必然的。

5. 常见问题与性能调优

5.1 高频踩坑问题速查表

我把这段时间遇到的高频问题整理成一张表，每个问题都标注了现象、根本原因和解决办法：

这里面最隐蔽的是第二个，空闲连接断开。MySQL默认wait_timeout是8小时，看起来够长，但如果网络环境里还有一层负载均衡器，它的空闲超时往往只有几分钟。Flink CDC长连接一旦空闲超过中间层阈值被断开，任务就报错。解决思路是开启连接保活机制，让连接一直处于活跃状态，或者和网络团队确认中间层的空闲超时时间。

5.2 性能瓶颈与优化思路

跑通只是第一步，数据量上来之后才是真正的考验。同步链路的性能瓶颈一般出现在三个位置。

第一是源库binlog的解析速度。Flink CDC连接器内部解析binlog是单线程的，在单个连接器实例下这块基本是瓶颈。如果单表写入量极大，可以按表拆分多个Flink CDC源，让每个任务的解析互不干扰。不推荐对单张表做并行解析，binlog的位点连续性决定了单表并行很难做。

第二是Flink算子间的数据传输序列化开销。打开Flink Web UI观察BackPressure指标，如果下游算子长期处于高背压状态，说明sink写入速度跟不上上游。这时候优先调整ClickHouse Sink的批次参数，把sink.batch-size往上调，批量越大写入吞吐越高，但延迟也会相应增大。这个要根据业务对实时性的要求权衡，我一般用1000到5000之间。

第三是ClickHouse自身的写入能力。ClickHouse适合批量写入，频繁的小批次写入会产生大量小分区，后台合并线程压力大，查询变慢。解决思路是不要让Flink的flush间隔太短，至少1秒以上，同时ClickHouse侧设置合理的分区键，避免每个批次都落到新分区。如果ClickHouse是集群模式，写入要尽量打散到每个节点，分布式表的写入配置也要单独调。

这几轮优化做完，我这边实际压测的结果是：单表千万级数据全量快照，加上持续写入压力，端到端延迟能控制在秒级以内，源库的CPU和IO开销几乎可以忽略。这个表现，定时批处理方案是完全给不了的。

注意：性能调优没有银弹，一定要先通过Flink Web UI定位瓶颈在哪个环节，再对症下药。盲目调大并发有时候反而会让源库和ClickHouse双方都压力倍增。

6. 实操心得与后续扩展

写到这里，最后分享几个只有真正跑过生产才会懂的经验。

第一件事，上线前一定要做故障演练。不是让任务跑通就收工，而是故意把Flink任务kill掉，等十分钟再重启，观察数据能否自动对齐。只有经历过一次真实的断点恢复，你才会明白Checkpoint目录、binlog保留时间、目标表去重字段这三者是怎么协同工作的。我见过太多团队，任务跑了一周都正常，第一次重启就丢数据，原因就是没验证过恢复路径。

第二件事，监控不能只看任务是否Running。Flink任务挂在那边不代表数据在流动。建议接入任务延迟指标，Flink CDC内置了Source的延迟监控，延迟一旦超过阈值就告警。ClickHouse侧也要监控表行数和最新数据时间，两边交叉验证，才能保证数据链路的健康。

第三件事，这套方案后续扩展空间很大。Flink CDC不止能同步MySQL，PostgreSQL、Oracle、SQL Server都支持，目标端也不止ClickHouse，Kafka、Doris、Iceberg都可以。如果以后要搭实时数仓，直接把CDC链路里的sink从ClickHouse换成Kafka，上游实时计算再接入Flink做流式ETL，整个架构就自然演进过去了。

我自己实际操作下来的体会是，数据同步这件事，工具选对了能让复杂度降一个量级。Flink CDC最大的价值，是让我不再花精力去处理"怎么把数据搬过去"的问题，而是把精力放在"搬过去之后怎么用"的问题上。这也是为什么我现在做类似需求时，会优先考虑这套方案而不是自己造轮子。希望这篇文章能帮你少走几个我刚才说的那些弯路。