最近有个准实时数仓的同步需求要把 MySQL 和 PostgreSQL 里的几张业务表实时汇到统一存储里。我一开始图省事打算用脚本轮询结果发现数据量一大根本撑不住而且字段变更、删除操作全都对不上最后还是老老实实回到 Flink CDC 这条路上。这次我选了 Apache Flink CDC 的 standalone 模式来做部署。原因很简单公司还没有现成的 Flink 集群YARN/K8s 又需要额外维护业务量还没到必须容器化的阶段先在一台机器上把链路跑通、验证数据准确性后续再平滑迁移到真正的大集群里。整个过程踩了不少坑从版本匹配到连接器类加载再到 PostgreSQL 的复制槽管理几乎每一步都有一堆隐藏细节。这篇文章我会完整还原自己的实际操作过程怎么选版本、怎么搭 standalone 环境、怎么写 Flink SQL 同时拉取 MySQL 和 PostgreSQL 的变更数据以及部署之后如何管理 Checkpoint 和处理那些常见的坑。适合那种手边没有现成 Flink 集群、想快速把 CDC 跑起来做数据验证的开发者也适合刚入门 Flink CDC、想看真实落地路径而不是只看官网 demo 的人。1. 部署前必须想清楚的几件事版本组合与运行模式1.1 版本搭配是第一道坎Flink 和 CDC Connector 的匹配矩阵Flink CDC 并不是一个独立的服务它本质上是一组连接器插件挂在 Flink 生态里。所以第一件事就是把 Flink 发行版和 CDC connector 的版本对应关系搞清楚。我见过太多人随便拿一个 Flink 2.x 和 CDC 2.x 拼在一起结果启动后各种ClassNotFoundException。以我本次使用的组合为例Flink 1.16.2 Flink CDC 2.4.x。这个组合在实际生产里用得很多也相对稳定。Flink 1.16 对应 Scala 2.12下载的时候要注意选带bin-scala_2.12的包。如果你用的是 Flink 1.15 或 1.17CDC connector 的编译兼容性会略有差异部分参数名也变了。Flink 版本推荐的 CDC connector 版本注意事项1.14.x2.2.x老项目常见但 checkpoint 锁粒度较粗1.15.x2.3.x需要关注 connector 的 jar 包命名1.16.x2.4.x我本次使用的组合稳定性不错1.17.x2.4.x 或更新官方文档里已放入新目录结构建议先读 release notes这里想多说一句不要只看大版本号还要看连接器 jar 包的完整文件名。Flink CDC 的发布包里分为flink-sql-connector-mysql-cdc-xxx.jar这种“胖包”以及flink-connector-mysql-cdc-xxx.jar这种“瘦包”。SQL Client 场景必须使用flink-sql-connector-前缀的胖包因为它把 Debezium、Kafka Connect API 等依赖都打进去了。瘦包更适合自己写 DataStream 程序时通过 Maven 管理依赖直接丢到 lib 目录里很容易因为缺失传递依赖而报错。1.2 为什么选 standalone 而不是 YARN 或 K8s很多教程一上来就让你开 YARN、搭 Hadoop然后本地可能根本没有集群卡在环境上。Flink standalone 模式虽然不能享受资源动态伸缩但对 CDC 这类“少量 source sink”的实时同步场景其实完全够用。standalone 的优点是部署简单角色清晰一个 JobManager 负责调度一个或多个 TaskManager 负责执行任务Web UI 直接可见作业状态。缺点是 JobManager 单点且没有自动故障重启的集群语义。所以我的建议是单机验证、学习原理、数据量不大的准生产环境用 standalone一旦有多个作业并行、需要 HA 或动态资源就赶紧迁到 YARN/K8s。另外要说清楚Flink 自带的 local 模式和 standalone 是有区别的。local 模式是在进程内起一个 MiniCluster常用于代码测试standalone 是独立进程可以持续运行作业也能通过flink run提交任务。CDC 同步这种长任务必须跑在 standalone 上而不是本地开发环境里跑一下就退出。1.3 机器资源与端口规划standalone 模式对机器的要求不高但如果你要同时跑 MySQL CDC 和 PostgreSQL CDC建议至少 4 核 8G。因为我实际测下来每个 CDC source 都会初始化快照线程和 binlog/LSN 读取线程TaskManager 默认只有 1 个 slot 时作业能跑但吞吐上不去。默认端口方面Flink Web UI 是 8081JobManager 的 RPC 是 6123。如果你机器上已经挂了其他服务记得改配置。还有一个极容易忽略的是 hostnameconf/flink-conf.yaml里的jobmanager.rpc.address默认是localhost如果你用主机名访问可能起不来。建议部署前直接统一改成 IP或者把 hostname 加进/etc/hosts免得来回折腾。2. 下载与安装把 Flink 和 CDC 连接器放到正确的位置2.1 获取 Flink 二进制包并解压这一步没什么技巧从 Flink 官网下载页找到对应版本用wget拉下来就行。注意选择flink-1.16.2-bin-scala_2.12.tgz不要选带src的源码包。wget https://archive.apache.org/dist/flink/flink-1.16.2/flink-1.16.2-bin-scala_2.12.tgz tar -zxvf flink-1.16.2-bin-scala_2.12.tgz mv flink-1.16.2 /opt/flink export FLINK_HOME/opt/flink export PATH$PATH:$FLINK_HOME/bin解压之后确认两个可执行文件的权限bin/start-cluster.sh和bin/sql-client.sh都在就行。2.2 获取 MySQL CDC 和 PostgreSQL CDC 的连接器 jarFlink CDC 连接器可以从 Maven 中央仓库直接下也可以去项目的 release 页面找。我需要的是两个wget https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-mysql-cdc/2.4.1/flink-sql-connector-mysql-cdc-2.4.1.jar wget https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-postgres-cdc/2.4.1/flink-sql-connector-postgres-cdc-2.4.1.jar把这两个 jar 放进$FLINK_HOME/lib/目录。这一步是很多人忽略的关键点SQL Client 启动时会扫描 lib 目录加载所有 jar所以连接器必须放在这里而不是用ADD JAR命令临时添加。ADD JAR在部分版本里只对当前 session 有效作业提交之后 TaskManager 未必能加载到。如果你用的是稍新一点的 CDC 版本连接器 jar 可能在lib/下还依赖flink-connector-base之类的其他 jar建议解压后检查一下lib目录把缺失的依赖一起放进去。我用的是 2.4.1 胖包实测没有额外依赖问题。2.3 启动 Flink 集群并验证进入$FLINK_HOME执行bin/start-cluster.sh等几秒钟用jps查看进程应该能看到两个 Java 进程StandaloneSessionClusterEntrypoint和TaskManagerExecutor。然后浏览器打开http://localhost:8081能看到 Flink Web UI 就说明集群起来了。这里提一个我踩过的坑如果jps只看到了StandaloneSessionClusterEntrypoint没有 TaskManager大概率是内存分配问题。默认taskmanager.memory.process.size是 1728M如果机器上可用内存不足TaskManager 会启动失败。这时候可以调低比如改成 1024M或者直接加大机器内存。3. 第一个实战MySQL CDC 单表同步到 Print Sink3.1 准备 MySQL 测试环境并开启 binlogMySQL CDC 的原理是伪装成 MySQL 的从库读取 binlog。所以你的 MySQL 必须开启ROW格式的 binlog否则 CDC 拿不到完整的前后镜像数据。我使用的是 MySQL 8.0 版本。修改/etc/my.cnf[mysqld] server-id 1 log_bin mysql-bin binlog_format ROW binlog_row_image FULL expire_logs_days 7这里每一个参数都有讲究server-id整个 MySQL 复制环境里必须唯一不能和已有的从库、其他 CDC 任务冲突。后面创建多个 CDC 任务时每个任务也必须分配不同的 server-id否则两个任务会互相踢掉对方的 binlog dump 连接。binlog_formatROWFlink CDC 依赖行级变更记录只有 ROW 格式才能看到每行的 before/after 镜像。binlog_row_imageFULL确保记录所有列的最大镜像而不是只记修改列否则 Debezium 解析出的字段会不完整。expire_logs_days控制 binlog 保留时长防止长时间不消费导致磁盘爆掉。修改后重启 MySQLsudo systemctl restart mysqld重启后验证一下SHOW VARIABLES LIKE log_bin; SHOW VARIABLES LIKE binlog_format;两条结果都符合预期后再创建用于 CDC 的账号CREATE USER cdc_user% IDENTIFIED BY cdc_pass; GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO cdc_user%; FLUSH PRIVILEGES;权限里的SELECT是为了读取快照REPLICATION SLAVE和REPLICATION CLIENT是为了建立 binlog dump 连接并获取服务端位点。很多教程只给了REPLICATION SLAVE结果作业报权限不足白折腾半天。3.2 创建测试表并插入数据我先建了一个测试库和一张用户表CREATE DATABASE testdb; USE testdb; CREATE TABLE t_user ( id INT PRIMARY KEY, name VARCHAR(50), age INT, created_at TIMESTAMP ); INSERT INTO t_user VALUES (1, Alice, 28, NOW()); INSERT INTO t_user VALUES (2, Bob, 32, NOW());3.3 在 Flink SQL Client 里建表、跑数据启动 SQL Clientbin/sql-client.sh然后挨个执行SET execution.checkpointing.interval 5s; CREATE TABLE mysql_user_source ( id INT, name STRING, age INT, created_at TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector mysql-cdc, hostname localhost, port 3306, username cdc_user, password cdc_pass, database-name testdb, table-name t_user, server-id 5400-5404 ); CREATE TABLE print_sink ( id INT, name STRING, age INT, created_at TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector print ); INSERT INTO print_sink SELECT id, name, age, created_at FROM mysql_user_source;执行完最后一行之后Flink 会启动一个持续运行的作业。你可以看到输出里先打印出已经存在的两条数据然后我再往 MySQL 里插入一条新数据INSERT INTO t_user VALUES (3, Charlie, 25, NOW());几乎一瞬间SQL Client 的终端里就会多出一行I(3, Charlie, 25, ...)。这就是 CDC 的增量捕获能力。继续试试更新和删除你会看到-U、U、-D这样的变化前缀分别对应 UPDATE 的旧镜像、新镜像和 DELETE 记录。如果你不想让作业一直刷屏可以在 Web UI 里找到对应作业点击 Cancel 让它停止。Print sink 本身没有外部存储只用于验证侵入性为零。3.4 理解 scan.startup.mode 和历史数据初始化上面这个例子默认行为是先做一次全量快照把表当前数据读出来然后在快照完成的位置开始追 binlog。这个模式叫initial。如果你的表非常大几亿行那初始化阶段会持续很长时间期间千万不要手动取消作业否则下一次从 Checkpoint 恢复时可能会重复扫描。如果你只想消费“从现在开始”的增量不关心历史数据可以把启动参数改成scan.startup.mode latest-offset这样连接器不会读取已有数据而是从 binlog 当前位点开始。这个模式适合那种“只要事后增量不希望全量刷一遍”的场景。反过来如果你想从某个具体时间的 binlog 开始回放可以用timestamp模式但需要额外配置scan.startup.timestamp-millis而且要求 binlog 文件的保留时间覆盖那个时间点否则会抛错。4. 第二个实战PostgreSQL CDC 同步与逻辑复制配置4.1 PostgreSQL 开启逻辑复制相关的配置PostgreSQL 的 CDC 底层用的是逻辑复制机制和 MySQL 的 binlog 原理不太一样。它通过解析 WAL预写日志中已提交的事务并以逻辑流的方式重建变更。要在 PostgreSQL 端支持逻辑复制首先要修改postgresql.confwal_level logical max_replication_slots 10 max_wal_senders 10 listen_addresses *其中wal_level logical是必须的它会让 WAL 中额外记录足够用于逻辑解析的信息。max_replication_slots决定了可以同时存在多少个复制槽max_wal_senders决定了允许多少 WAL 发送进程。如果你是刚装的 PostgreSQL默认值通常是wal_level replica、max_replication_slots 10、max_wal_senders 10那只需要把wal_level改成logical然后重启。验证配置生效可以在 psql 里执行SHOW wal_level; SHOW max_replication_slots;4.2 创建复制账号和 PublicationPostgreSQL 的逻辑复制有“订阅端”和“发布端”的概念Flink CDC 在这里扮演的是订阅端所以我们需要在 PostgreSQL 里先创建一个发布端也就是 Publication。创建用户并授复制权限CREATE USER cdc_user WITH PASSWORD cdc_pass; ALTER USER cdc_user WITH REPLICATION LOGIN; GRANT CONNECT ON DATABASE testdb TO cdc_user; GRANT SELECT ON ALL TABLES IN SCHEMA public TO cdc_user;然后在要同步的数据库里创建发布\c testdb CREATE PUBLICATION dbz_publication FOR ALL TABLES;这里我用了FOR ALL TABLES意思是这个库下所有表都参与发布。Flink CDC 的 PostgreSQL connector 默认会去寻找名为dbz_publication的发布所以你最好用这个名字省得后面还要在 WITH 参数里手动指定。当然你也可以自定义比如CREATE PUBLICATION my_pub FOR TABLE public.t_user;对应的 Flink SQL 里就要写publication.name my_pub。4.3 用 Flink SQL 创建 PostgreSQL CDC source我同样建一张测试表并插入数据CREATE TABLE IF NOT EXISTS public.t_user ( id INT PRIMARY KEY, name VARCHAR(50), age INT, created_at TIMESTAMP ); INSERT INTO public.t_user VALUES (1, Alice, 28, NOW()); INSERT INTO public.t_user VALUES (2, Bob, 32, NOW());然后到 Flink SQL Client 里执行CREATE TABLE pg_user_source ( id INT, name STRING, age INT, created_at TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector postgres-cdc, hostname localhost, port 5432, username cdc_user, password cdc_pass, database-name testdb, schema-name public, table-name t_user, slot.name flink_slot_1, decoding.plugin.name decoderbufs ); INSERT INTO print_sink SELECT id, name, age, created_at FROM pg_user_source;这里有两个参数要特别注意slot.name复制槽名称。同一个 slot 在同一时间只能被一个订阅端消费所以多个 PostgreSQL CDC 任务必须使用不同的 slot 名称否则后启动的作业会一直等待甚至报错。decoding.plugin.name逻辑解码插件。默认值是decoderbufs需要 PostgreSQL 端安装decoderbufs插件。如果没装也可以换成wal2json或pgoutput但要确保 PostgreSQL 对应插件存在。如果你在启动时报错找不到org.postgresql.Driver说明运行的集群里缺少 PostgreSQL JDBC 驱动。虽然flink-sql-connector-postgres-cdc胖包里理论上应该包含但不同版本可能没有打进去。解决办法是下载postgresql-42.x.x.jar也放进 Flink 的 lib 目录然后重启 SQL Client。4.4 复制槽不释放导致 WAL 膨胀的问题使用 PostgreSQL CDC 时最容易被忽视的就是复制槽的生命周期。Flink 作业运行时PostgreSQL 会为它保留一个复制槽并且所有未被消费的 WAL 都不会被清理。如果作业异常退出、复制槽没有被释放而你的max_wal_senders和磁盘空间允许的话PostgreSQL 会很快被 WAL 文件塞满。排查方法是在 PostgreSQL 里查询SELECT slot_name, active, restart_lsn, confirmed_flush_lsn FROM pg_replication_slots;如果某个 slot 的active是false但还保留着那说明作业已经停了但槽还在。可以手动清理SELECT pg_drop_replication_slot(flink_slot_1);但在清理前一定要确认这个作业已经不再需要从该位点恢复否则会丢失未消费的变更。我的经验是给每个 Flink 作业固定一个明确的slot.name作业停止后立刻检查是否还残留。如果只是临时验证建议在作业取消后马上 drop养成习惯不然哪天磁盘报警就晚了。5. 运行期状态管理Checkpoint 对 CDC 任务的意义5.1 为什么 CDC 任务必须开启 CheckpointFlink CDC 连接器内部维护了很多状态比如 binlog 位点、schema 历史、未完成的快照分片等。如果不开启 Checkpoint作业一旦重启Debezium 引擎会丢失当前读取到的 binlog offset导致两个严重后果一是从旧位点开始重复消费大量历史变更二是可能错过停机期间的增量数据。Flink 的 Checkpoint 机制会定期把状态快照保存到持久化存储。对于 CDC 任务来说最直观的表现是当作业恢复时MySQL 连接器会从 Checkpoint 里记录的 binlog position 继续读不需要重新做全量快照。所以我在每次创建同步作业时都会在 SQL Client 里先执行SET execution.checkpointing.interval 5s; SET execution.checkpointing.mode EXACTLY_ONCE; SET execution.checkpointing.min-pause 500ms;间隔设多少合适我一般用 5-10 秒。太短的间隔会把状态写入频繁化浪费 IO太长的间隔则会让恢复时可能重复处理最近几分钟的数据。同步场景下EXACTLY_ONCE配合下游幂等写入基本可以实现最终一致。5.2 状态后端选型从 HashMap 到 RocksDB默认情况下Flink 的状态后端是 HashMapStateBackend状态保存在 TaskManager 的堆内存里。CDC 大表快照阶段特别是千万级以上的表状态很容易涨到几个 GB堆内存根本扛不住。推荐直接改成 RocksDB 状态后端并配合文件系统作为 Checkpoint 存储SET state.backend.type rocksdb; SET state.checkpoint-storage filesystem; SET state.checkpoints.dir file:///opt/flink/checkpoints;RocksDB 的缺点是本地磁盘 IO 会高一些但换来了稳定的内存占用。对单机 standalone 来说只要磁盘有空间这套组合可以支撑比较大的同步任务。你可能还想把增量 Checkpoint 打开这个在较新的 Flink 版本里是SET execution.checkpointing.incremental true;增量 Checkpoint 只保存自上次 Checkpoint 以来发生变化的状态部分可以明显缩短大状态场景下 Checkpoint 的耗时。但注意如果后续要从旧版 Flink 迁移状态增量 Checkpoint 文件可能不兼容升级前最好手动触发一次全量 savepoint。5.3 savepoint 与手动恢复除了自动 Checkpoint运维上更重要的是学会手动保存 savepoint。比如你要调整 SQL 里的某个参数或者升级连接器版本总不能每次都从零开始重新全量同步。手动触发 savepoint 的命令是bin/flink savepoint jobId file:///opt/flink/savepoints恢复时需要携带 savepoint 路径重新提交作业bin/flink run -s file:///opt/flink/savepoints/savepoint-xxxx \ -d -c 主类或使用sql应用 ...如果是 SQL 作业恢复会复杂一些。我通常的做法是把原来的 DDL 语句放进一个 SQL 文件然后用sql-client.sh -f xxx.sql执行。不过 SQL Client 提交的作业无法直接用flink run指定 savepoint 恢复这一点有坑。更稳妥的方案是直接用 Flink DataStream API 写提交程序或者使用 Flink CDC 的SqlClient配合flink run的 application 模式。如果你只是做简单验证我的建议是不要纠结于 SQL Client 的 savepoint 恢复而是记录好作业的 Checkpoint 目录作业挂掉后直接重启集群Flink 会自动从最近一次 Checkpoint 恢复。这需要flink-conf.yaml里配置execution.savepoint.path: file:///opt/flink/savepoints/latest这样重启后会自动尝试加载稳定省心。6. 踩坑实录我在 standalone 部署中遇到的典型问题6.1 SQL Client 启动后报找不到连接器类最常见的现象是执行CREATE TABLE ... WITH (connectormysql-cdc)时报java.lang.ClassNotFoundException: com.ververica.cdc.connectors.mysql.MySqlSource我第一次遇到时第一反应是版本不匹配后来检查发现连接器 jar 根本没有被加载。原因是 SQL Client 启动时会扫描lib目录如果 jar 放到了opt目录或临时目录自然找不到。排查其实就两步ls $FLINK_HOME/lib确认 jar 文件名和大小注意大小。胖包通常在 80MB 以上瘦包只有几 MB。如果 jar 存在还是报错把$FLINK_HOME/log/flink-*-taskexecutor-*.log打开看底层异常是不是缺失其他依赖。还有一种情况是你同时在 lib 里放了 MySQL 官方 JDBC 驱动和一个老版本的 CDC jar两个 jar 里的类路径冲突也会导致加载异常。解决办法是把多余的 jar 清理掉只保留一个flink-sql-connector-mysql-cdc胖包因为它已经包含了 JDBC 驱动。6.2 MySQL CDC 作业启动后反复断开或读不出增量这个坑的典型症状是作业能启动历史数据也能读出来但一运行几分钟就报错ERROR: Got error reading packets from server: slave is not configured or the driver cannot connect to the specified server多半是 server-id 冲突。MySQL 会要求每个 binlog dump 客户端使用唯一的 server-id如果你在多个任务里都写了server-id 5400那么后一个连接会把前一个连接踢掉导致通道频繁重置。我的处理方法是每个任务都配置一段 server-id 范围而不是单个值server-id 5400-5404范围里的每一个数字都可以分配给并行子任务如果 source 并行度是 1范围取一个就够但为了避免和别的任务撞上尽量分配一段独立区间。另外如果你的 MySQL 是 5.6 或更老的版本某些版本的 binlog 解析会产生兼容问题。建议统一用 5.7 或 8.0省事很多。6.3 PostgreSQL 复制槽不释放导致 WAL 把磁盘撑满这是我这次部署里最惊险的一次。因为中途反复调试 SQL、取消作业PostgreSQL 里残留了多个复制槽pg_wal目录一路涨到几十 GB差点把服务器根分区写满。现象是 PostgreSQL 日志里出现ERROR: requested WAL segment ... has already been removed这说明 WAL 已经被清理但某个复制槽因为长期不消费导致 PG 认为还需要保留更早的日志。实际处理方式分两步把所有不再使用的 slot 全部删除SELECT pg_drop_replication_slot(slot_name) FROM pg_replication_slots WHERE active false;对仍被占用的 slot确认对应的 Flink 作业是否还在。如果作业本身已经挂了但 slot 还显示 active大概率是 TaskManager 里还有残留连接重启 Flink 集群后再检查。后来我养成了习惯每个 PostgreSQL CDC 任务在启动前先检查pg_replication_slots里有没有同名 slot有的话先统一清理一次再提交作业。这样能避免很多莫名其妙的问题。6.4 时间字段和时区偏差使用 MySQL CDC 时如果下游清洗后发现时间数据比业务实际时间少了 8 小时大概率是连接器在解析timestamp类型时按照 UTC 处理了。Flink CDC 的 Debezium 内部会把数据库时间字段转换成 UTC 的字符串。如果你在 Flink SQL 里定义了TIMESTAMP(3)最终打印出来的时间会和原库相差一个时区偏移。解决办法是给连接器配置时区相关参数server-time-zone Asia/Shanghai, debezium.event.processing.timezone Asia/Shanghai第一个参数是连接器读取 MySQL binlog 时的会话时区第二个参数是 Debezium 引擎处理事件流时的时间转换规则。两个都设置成目标时区通常就能把时间修正回来。PostgreSQL 侧相对简单因为逻辑复制输出里的时间戳一般带时区信息。如果你在 DDL 里把字段声明为TIMESTAMP WITH LOCAL TIME ZONEFlink 会自动转换成任务本地时区不容易出现那种整小时偏移的问题。最后说一点个人的体会这套 standalone 部署跑通后我最大的感受是Flink CDC 强大归强大但每一个参数背后都有一条真实的故障链路。版本不匹配会让你怀疑人生server-id 冲突会让你找不到原因复制槽不清理会让磁盘悄悄爆炸。与其把这些问题当成“偶然的坑”不如在写第一条同步 SQL 之前就按文档里的推荐配置做好。先用小表、纯 Print sink 方式验证连通性再逐步加入生产级的 Checkpoint、状态后端和监控这种梯度推进方式比一上来就想同步全库要稳得多。如果后续业务量上来了我会考虑把 standalone 集群替换成 YARN 上的 session 集群并把多个同步作业统一交给调度平台管理。但底层的 Flink CDC SQL 写法基本不会变这也是它最值得投入学习的地方。希望这篇操作笔记能帮你少跳几个坑。