Flink CDC Standalone下的MySQL与PostgreSQL同步

如果你和我一样，摊上过那种“业务库每隔几分钟就要往数仓同步一批数据”的活，最初多半是从写定时脚本开始的。全量抽数一多就慢，后来换成定时增量，又要自己维护时间戳字段，DML 和表结构变化都顾不全。直到我在一个项目里试了 Apache Flink CDC 这套方案，才真正把链路理顺。Flink CDC 最大的价值，是它把数据库日志解析这件事变成了 Flink 里的普通数据源连接器，你可以直接在 SQL 里建一张“会变化的表”，然后像普通表一样去查询、过滤、关联、写目标端。本文就基于 Flink standalone 模式，把 MySQL CDC、PostgreSQL CDC 的部署、配置和使用方法完整跑一遍。整篇文章按“部署—源表—同步—排障”的顺序来，适合刚接触实时同步、手头又只有几台服务器的小团队，也适合想先在本机把链路跑通再上生产的同学。

1. 整体思路与方案选型

1.1 standalone 模式适合什么场景

Flink standalone 是 Flink 自带的最小集群形态，一个 JobManager 加若干个 TaskManager，不依赖 YARN、Kubernetes、Mesos 这类资源管理系统。它的好处非常直观：下载解压就能用，出了问题不会牵扯到调度器的概念，日志和 Web UI 都直来直去。

我在实际项目里选择 standalone，通常是这三种情况之一。第一种是本地或测试环境，只想验证一条 CDC 同步链路能不能走通，没必要为一个实时任务搭整套资源管理平台。第二种是中小规模数据量，任务数量和并发都不高，standalone 自带几台机器完全撑得住。第三种是团队里还没有专门维护大数据基础设施的人，standalone 的运维半径最小，一个人也能搞定。

但它也确实有限制。standalone 模式下 JobManager 默认没有热备机制，TaskManager 挂了也不会自动拉起，资源之间没有严格隔离，多个作业挤在一起可能互相影响。所以我一般会提醒一句：生产环境如果不是特别小的规模，建议把同样的 SQL 和作业逻辑往托管模式上迁移，好在 Flink CDC 的 DDL 和作业代码并不绑定部署方式，迁移成本很低。

1.2 Flink CDC 的数据链路是怎么工作的

Flink CDC 连接器内部遵循的是“先快照、后增量”的思路。你可以理解成给数据库表先拍一张全景照片，然后再持续回放之后发生的每一帧变化。这个设计最大的价值是：同步任务启动时不需要停业务，也不需要你去记录当前数据到哪个位置了，连接器会自动完成一致性快照，并在快照结束时无缝切换到日志解析。

具体到实现层面，MySQL CDC 依赖 binlog，PostgreSQL CDC 依赖逻辑复制和 WAL，连接器会把这些日志流转换成一个动态表。对 Flink 来说，它看到的就是一张会持续变化的表，你在这张表上写一条 INSERT INTO，就完成了从源库到目标库的同步。

这里还有一个很重要的机制是 checkpoint。日志解析的当前位点会作为 Flink 状态的一部分，随着 checkpoint 周期性落盘。任务异常重启后，Flink 从最近一次 checkpoint 恢复，连接器从对应的日志位点继续消费，这样就做到了不大面积丢数。配合下游幂等或主键去重，整体可以达到比较高的数据一致性。

1.3 版本选型：先定 Flink，再选连接器

Flink CDC 和 Flink 主版本之间的匹配关系，基本是初学者踩坑重灾区。我见过太多次“Flink 1.15 里放了新版 connector，TaskManager 疯狂报 ClassNotFound”的情况。

我常用的稳定组合大致是这样：

我现在的习惯是优先选 Flink 1.18 配 3.1 左右的连接器，这个阶段的功能和参数命名相对成熟，网上能搜到的踩坑帖也比较多，出了问题容易找到参照。另外连接器 Jar 包通常区分 Scala 2.12 / 2.13，下载和引入依赖时要注意和 Flink 发行版保持一致。

2. Flink standalone 集群的部署与准备

2.1 部署前的系统准备

Flink 依赖 JDK，我一般直接装 JDK 1.8 或 11，版本太老或太新都容易遇到兼容问题。装好后先确认 java -version 能正常输出，再下载对应版本的 Flink 二进制包。

bash复制sudo mkdir -p /opt/flink
sudo tar -zxvf flink-1.18.1-bin-scala-2.12.tgz -C /opt/flink
export FLINK_HOME=/opt/flink/flink-1.18.1
export PATH=$FLINK_HOME/bin:$PATH

解压之后重点看几个目录：bin 放启动脚本，conf 放配置文件，lib 是 Flink 自身和依赖 Jar 的加载目录，plugins 是扩展插件目录。以后排查依赖问题，基本就在 lib 里翻。

之所以强调目录结构，是因为 standalone 模式下所有 TaskManager 都是从同一个 FLINK_HOME 启动的，你把连接器 Jar 放进 lib，所有节点都能加载到。这一点在后面配 SQL Client 时尤为重要。

2.2 关键配置与启动步骤

配置文件在 conf/flink-conf.yaml，我通常先改三个核心参数：

yaml复制jobmanager.memory.process.size: 2048m
taskmanager.memory.process.size: 2048m
taskmanager.numberOfTaskSlots: 4

jobmanager.memory.process.size 是 JobManager 进程的整体内存，包括 JVM 堆、元空间、直接内存等；taskmanager.memory.process.size 同理，指的是每个 TaskManager 的总内存，而不是堆内存。Flink 在 standalone 模式下会在系统内存之外预留一部分给 JVM 开销，所以不能把机器内存全填进去，不然启动后直接 OOM。

taskmanager.numberOfTaskSlots 决定一个 TaskManager 能同时跑多少个 Slot。我一般先按 CPU 核心数设置，测试环境设 2 到 4 都可以，不要贪多。同一台机器同时跑多个 CDC 作业时，Slot 数太少会导致任务排队等待。

如果你有多台机器，需要在 conf/workers 文件里把各节点的主机名或 IP 写进去，每行一个。单机环境默认写 localhost 即可。

启动集群：

bash复制$FLINK_HOME/bin/start-cluster.sh
jps

正常情况下，jps 能看到 StandaloneSessionClusterEntrypoint 和 TaskManagerRunner 两个进程。再打开 http://localhost:8081，进入 Web UI 确认 Available Slots 数量，如果显示为 0 或报错，优先检查内存配置和防火墙。

2.3 把 SQL Client 需要的 Jar 放对位置

Flink standalone 能跑起来只完成了一半，我们还要让 SQL Client 能读到 CDC 连接器。我自己最常用的做法是：把 flink-sql-connector-mysql-cdc-*.jar、flink-sql-connector-postgres-cdc-*.jar、flink-connector-jdbc-*.jar 全部放到 $FLINK_HOME/lib 下，然后重启集群。

这一步背后的原因是：SQL Client 只是一个提交客户端，真正的作业会被分发到 TaskManager 上执行。如果你只在 SQL Client 启动时用 -j 临时指定 Jar，客户端本地可能没问题，但作业跑起来后 TaskManager 找不到类，就会报各种 ClassNotFound。放进 lib 是为了让每个节点都“自带干粮”。

放好 Jar 后启动 SQL Client：

bash复制$FLINK_HOME/bin/sql-client.sh

进入 REPL 后输入：

sql复制SET 'sql-client.execution.result-mode' = 'tableau';

这是把结果渲染成更适合终端阅读的表格模式，实测下来体验好很多。到这里，standalone 集群和 SQL Client 环境就准备好了。

注意：更换或新增 Jar 后一定要重启集群和 SQL Client。单独重启 JobManager 不够，TaskManager 进程内存里已经加载的类不会自动更新，这也是很多人踩过的一个隐藏坑。

3. MySQL CDC 实操

3.1 提前改好 binlog 与账号权限

MySQL 的 CDC 同步依赖 binlog，所以源库必须要开启 binlog，而且格式必须是 ROW。ROW 格式记录的是每一行数据实际变化的前后镜像，STATEMENT 格式只记录 SQL 语句，在 UPDATE 和 DELETE 场景下没法可靠还原数据。

先检查当前状态：

sql复制SHOW VARIABLES LIKE 'log_bin';

如果没开，修改 my.cnf：

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

binlog_row_image = FULL 保证 UPDATE 事件里同时包含更新前后的完整行数据，这是 Flink CDC 能正确生成 before / after 结构的关键。expire_logs_days 建议设 7 天以上，同步任务如果长时间停机，过期 binlog 会被清理，任务恢复后找不到起始位点就麻烦了。

改完配置需要重启 MySQL。如果用的是云数据库，通常要在控制台开启 binlog 或多了一个“参数组”入口，操作路径稍有不同，但原理一样。

接下来创建同步账号并授权：

sql复制CREATE USER 'flinkuser'@'%' IDENTIFIED BY '你的密码';
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'flinkuser'@'%';
FLUSH PRIVILEGES;

这几个权限各有用途。SELECT 用于快照阶段读全量数据；REPLICATION SLAVE 和 REPLICATION CLIENT 用于请求 binlog 和读取主库状态；RELOAD 和 SHOW DATABASES 是连接器做元数据检查时需要的。不要只给一个 SELECT，很多权限不足的报错要到分布式任务运行一段时间后才冒出来，排查起来很麻烦。

3.2 用 Flink SQL 注册 MySQL CDC 源表

环境好了，数据库也开了 binlog，接下来就是在 SQL Client 里建表。我以一个用户表 users 为例：

sql复制CREATE TABLE mysql_users (
  id BIGINT,
  name STRING,
  email STRING,
  update_time TIMESTAMP(3),
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = '127.0.0.1',
  'port' = '3306',
  'username' = 'flinkuser',
  'password' = '你的密码',
  'database-name' = 'app_db',
  'table-name' = 'users',
  'scan.startup.mode' = 'initial',
  'server-time-zone' = 'Asia/Shanghai'
);

scan.startup.mode 是最需要理解的参数，它决定任务启动时从哪里开始读：

server-time-zone 直接关系到时间字段是否正确。MySQL 驱动在解析 DATETIME / TIMESTAMP 时会依赖时区参数，如果你的 Flink 集群和业务库不在同一个时区，时间很容易偏移 8 小时，这个参数建议显式设置。

database-name 和 table-name 都支持正则表达式，比如写 'app_.*' 就能匹配多个库表。如果一张表有多个主键或联合主键，在源表 DDL 里声明 PRIMARY KEY 时要把字段都列全，连接器在事件路由时非常依赖主键。

建完表后可以直接验证：

sql复制DESCRIBE mysql_users;
SELECT COUNT(*) FROM mysql_users;

能查出数据，说明快照链路已经通了。

3.3 写一个最简单的全量加增量同步作业

测试阶段我一般先建一张 print 结尾的临时表，专门用来观察事件流。connector = 'print' 会把每一条记录的变化都打印到 TaskManager 日志里，虽然不落地，但排查问题非常直观。

sql复制CREATE TEMPORARY TABLE user_sink (
  id BIGINT,
  name STRING,
  email STRING,
  update_time TIMESTAMP(3),
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'print'
);

INSERT INTO user_sink
SELECT id, name, email, update_time FROM mysql_users;

提交作业后，去源库做一次 UPDATE，再回到 SQL Client 看输出，能明显看到 -U 和 +U 两条记录，这就是 CDC 把更新前和更新后的行都推出来了。

实际业务里当然是要写进目标端。我用 JDBC sink 举例：

sql复制CREATE TABLE ods_users (
  id BIGINT,
  name STRING,
  email STRING,
  update_time TIMESTAMP(3),
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://127.0.0.1:3306/ods_db?useSSL=false',
  'table-name' = 'ods_users',
  'username' = 'ods_writer',
  'password' = '你的密码',
  'sink.buffer-flush.max-rows' = '1000',
  'sink.buffer-flush.interval' = '1s'
);

INSERT INTO ods_users
SELECT id, name, email, update_time FROM mysql_users;

jdbc sink 默认是批量 upsert，主键冲突时会转成 INSERT ... ON DUPLICATE KEY UPDATE。所以目标表必须有主键，且主键语义要和源表一致，否则 UPDATE 事件来了之后目标表会同时出现新旧两行，下游查数就乱了。

flush.max-rows 和 flush.interval 是吞吐和延迟的平衡点，测试阶段用默认值就行，线上按数据量调。这里更推荐直接用支持主键语义的 sink，比如 Doris、StarRocks、Kafka 落到下游再处理，JDBC sink 适合小表或低频场景。

3.4 checkpoint、任务停止与恢复

写完同步作业后，建议在提交前先打开 checkpoint：

sql复制SET 'execution.checkpointing.interval' = '60s';

checkpoint 是 Flink CDC 增量位点的“保险丝”。作业运行过程中，连接器消费到的 binlog 位点会随着 checkpoint 周期写入状态后端。任务挂了以后，只要从最近一次 checkpoint 恢复，就能从接近挂掉时刻的位点续读，不会从头再跑。

在 SQL Client 里停止作业时，也要注意方式。直接用 Ctrl+C 退出客户端而不主动触发 stop，作业的位点信息可能只停留在最近一次 checkpoint，丢失最近 60 秒的增量。我希望你记住的习惯是：先执行 STOP 语句触发 savepoint，再退出 SQL Client。如果作业已经跑在集群上，可以通过 Web UI 的 Cancel with Savepoint 操作，记录好 savepoint 路径，下次提交时用 -s 参数恢复。

这里要特别提醒一点：Flink CDC 的 DDL 自动同步能力在 2.x 阶段还比较弱。源表新增字段这类结构变更，连接器不会自动把新字段推给下游，通常需要你手动改目标表，再重启作业。所以线上表结构变更要有流程管控，不能直接在源库执行 online DDL 就完事。

4. PostgreSQL CDC 实操

4.1 PostgreSQL 侧准备：wal_level 和账号权限

PostgreSQL CDC 依赖的是逻辑复制，前提是数据库把 wal_level 设置成 logical。逻辑复制要求 WAL 里保留足够的信息，让解码模块能还原出每一行变更，这比物理复制要“重”一点，但对 Flink CDC 来说是必须的。

修改 postgresql.conf：

ini复制wal_level = logical
max_replication_slots = 10
max_wal_senders = 10

max_replication_slots 决定最多能同时存在多少个复制槽，Flink CDC 作业会默认占用一个；max_wal_senders 决定最多能有多少个 WAL 发送进程。这两个值建议至少保留 5 到 10 个的余量，因为除了 Flink，可能还有别的逻辑复制任务。

改完配置必须重启 PostgreSQL，wal_level 不支持在线修改。

然后创建账号：

sql复制CREATE USER flinkuser LOGIN REPLICATION PASSWORD '你的密码';
GRANT CONNECT ON DATABASE app_db TO flinkuser;
GRANT USAGE ON SCHEMA public TO flinkuser;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO flinkuser;

REPLICATION 权限是逻辑复制账号必须的，它允许账号发起 WAL 流复制请求。官方文档反复强调，生产环境不要用超级账号做 CDC 同步，最小权限账号出问题后也方便定位。

PostgreSQL 10 以上版本自带了 pgoutput 逻辑解码插件，理论上不需要额外安装其他解码插件。早期很多文档会教你编译安装 wal2json，但我实测下来新版本直接用 pgoutput 更省事，Flink CDC 连接器也默认优先使用 pgoutput。

4.2 通过 Flink SQL 注册 PG CDC 源表

建表语句和 MySQL 侧非常相似，但多了一个 slot.name 参数：

sql复制CREATE TABLE pg_orders (
  order_id BIGINT,
  user_id INT,
  amount DECIMAL(10, 2),
  create_time TIMESTAMP(3),
  update_time TIMESTAMP(3),
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '127.0.0.1',
  'port' = '5432',
  'username' = 'flinkuser',
  'password' = '你的密码',
  'database-name' = 'app_db',
  'schema-name' = 'public',
  'table-name' = 'orders',
  'slot.name' = 'flink_cdc_slot',
  'decoding.plugin' = 'pgoutput'
);

slot.name 是复制槽的名字，在一个 PostgreSQL 实例里必须全局唯一。如果之前有作业异常退出，复制槽可能还残留着，再次提交作业时会报“replication slot already exists”。这时候需要手动清理：

sql复制SELECT pg_drop_replication_slot('flink_cdc_slot');

清理操作要在没有其他任务占用该复制槽时执行，先停止 Flink 作业，再执行清理，否则会把正在跑的同步任务切断。

decoding.plugin 我建议固定写 pgoutput，它是 PostgreSQL 原生插件，和内置逻辑复制的兼容性最好。如果场景里有特殊的数据类型解码需求，再考虑 wal2json，但常规业务表用 pgoutput 就够了。

建完表后同样可以做一次快速验证：

sql复制SHOW wal_level;
SELECT slot_name, active FROM pg_replication_slots;

active 为 true 说明连接器已经在消费 WAL 了。

4.3 测试 UPDATE 和 DELETE 的增量事件

PG 链路的数据变更事件和 MySQL 一样，在 print sink 下直接可见。我建议在源库执行这样一组操作：

sql复制UPDATE orders SET amount = amount + 10 WHERE order_id = 100;
DELETE FROM orders WHERE order_id = 101;

回到 SQL Client 的 result 界面，能看到类似这样的变化流：

text复制+I[100, 88, 120.50, ...]
-U[100, 88, 110.50, ...]
+U[100, 88, 120.50, ...]
-D[101, ...]

-U 和 +U 配对表示一次更新，-D 表示删除。这个现象在 MySQL 上也一样，理解它有助于排查下游数据多一行、少一行的问题。

另外建议设置心跳参数：

sql复制SET 'heartbeat.interval.ms' = '5000';

PostgreSQL CDC 连接器如果长期空闲，WAL 复制槽的推进可能不够及时，导致源库 WAL 文件堆积。心跳事件会定期产生一个伪变更，促使位点持续推进。这个参数对长事务、低频业务表尤其重要。

4.4 和 MySQL 链路放在一起对比

两条链路我实际都跑过，差异主要集中在源端准备和参数调优上。下面这张表是我自己整理的心得：

server-time-zone 和 timestamptz 是两套时区坑里最容易踩的。MySQL 侧不指定时区，时间字段可能整体偏移；PG 侧如果字段是 timestamptz，连接器默认按 UTC 转成 Flink 的 TIMESTAMP_LTZ，你本地看到的可能是标准时间。解决方法是把源表字段声明为 TIMESTAMP，或者在查询时用 CAST(create_time AS TIMESTAMP) 做一次转换。

5. 高频问题与排查记录

5.1 版本不匹配导致的 ClassNotFound

报错类型通常是 NoClassDefFoundError 或 TableException: Could not find a valid factory，出现这两个词，先别急着看代码，直接去 lib 目录检查 Jar 包。常见原因是 lib 里同时放了两个版本，或者连接器版本和 Flink 主版本完全不匹配。

我的排查顺序是：先 ls $FLINK_HOME/lib | grep cdc 看有哪些 Jar；再用 unzip -p xxx.jar META-INF/MANIFEST.MF 看版本号；最后对照第 1.3 小节的版本匹配表确认。确认后删掉多余 Jar，重启集群，问题基本能解决。

5.2 MySQL 全量快照阶段锁表和超时

Flink CDC 做全量快照时，默认会把表按主键切成多个 chunk 并行读取。这个机制让 MySQL 不需要像老版 Sqoop 一样一把梭全表锁住，但如果大表没有主键，或者主键分布极不均匀，chunk 切分时依然可能长时间持有读锁，导致业务库报警。

遇到这个问题，优先给源表加主键或唯一索引；确实没法改表结构的，可以在连接器参数里指定一个非空唯一列：

sql复制'scan.incremental.snapshot.chunk.key-column' = 'id'

同时把 chunk 调小，减少单次持锁时间：

sql复制'scan.incremental.snapshot.chunk.size' = '4096'

如果快照阶段频繁报连接超时，检查 connect.timeout 和 connect.max-retries，这两个参数在连接器到数据库之间网络不稳时非常有用。

5.3 PostgreSQL 复制槽导致 WAL 无限增长

PG 的复制槽像一根“钉子”，会把 WAL 牢牢钉在磁盘上，直到消费端消费到那个位置。如果 Flink 作业挂了或者被强制取消，复制槽还残留，WAL 只会越来越多，最后把磁盘撑满。

排查命令：

sql复制SELECT slot_name, active,
       pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn) / 1024 / 1024 AS lag_mb
FROM pg_replication_slots;

active 为 false 且 lag_mb 越来越大，基本可以确定是死槽。处理方式是先确认没有作业在消费，然后：

sql复制SELECT pg_drop_replication_slot('flink_cdc_slot');

为了避免以后再踩，建议监控 pg_replication_slots 的 active 状态，并设置心跳参数持续推进位点。

5.4 时区与时间字段漂移

MySQL 的时区问题通常表现为：源库字段值明明是 12:00，同步到目标后变成 04:00 或 20:00。原因是连接器没有拿到正确的 server-time-zone，JDBC 驱动用默认时区解析了时间。解决办法就是给源表 DDL 里加：

sql复制'server-time-zone' = 'Asia/Shanghai'

PG 侧的问题大多出现在 timestamptz 字段。连接器按 UTC 解析后转成 TIMESTAMP_LTZ，展示层如果按本地时区渲染，就和源库对不上。我的处理方式是把目标字段声明为 TIMESTAMP(3)，并在 SELECT 里做一个显式转换，避免隐式时区问题。

5.5 无主键表的行为差异

无主键表在 CDC 链路里会非常难受。MySQL 侧没有主键时，连接器只能按 row 拼接标识，UPDATE 和 DELETE 事件的语义不如有主键时干净。PG 侧同样如此，逻辑复制事件在路由到目标行时找不到主键，会导致重复或丢失。

如果源表确实没有主键，我的建议是：第一步，推动业务侧加主键或唯一索引，这是根治方案；第二步，在源表 DDL 里用 PRIMARY KEY (col) NOT ENFORCED 声明一个逻辑主键，Flink SQL 会按这个 key 路由事件，但它不会检查源表物理主键是否存在，所以一定要选一个业务上真的唯一的列，否则会出错。

5.6 排障速查表

这些坑我基本都踩过一遍，经验是：排障顺序永远是“版本—权限—位点—参数”，不要一上来就去翻业务逻辑。

6. 关于这套链路的一些个人体会

走完这一整套流程，我最大的体会是：Flink CDC 的门槛其实不高，真正费精力的是版本组合和源端准备。几次排到深夜的问题，最后都发现只是 lib 里多了一个老 Jar，或者 PostgreSQL 复制槽没有及时清理。所以我的习惯是，任何新环境第一次跑 CDC，先用一张小表把“全量同步—增量变更—任务重启恢复”三个动作完整走一遍，确认没问题再上真实表。

另外建表只是开始，后续的维护才是大头。源表加字段、账号权限被回收、复制槽残留，这些不会在第一天出现，但三周后一定会冒出来。建议把源表和目标表的 DDL 变更纳入流程管理，至少保留一份审计记录，出问题的时候才能快速定位是谁改了什么。

最后再分享一个小技巧：用脚本封装 SQL Client 提交。把建表 DDL 和 INSERT 语句写进一个 .sql 文件，再用 sql-client.sh -f xxx.sql 方式提交，比在交互式 REPL 里一条条敲更容易复用，也方便接入告警。我现在的习惯是把这类脚本放进代码仓库，每次要起新同步任务时，直接复制改表名和参数就能跑。这套玩法很朴素，但真的省心。