开头部分：
手头有业务需要把 MySQL 和 PostgreSQL 的增量数据实时同步到下游，或者做实时数仓的 ODS 层接入，最先想到的方案大概率就是 Debezium，但真要快速把 CDC 能力吃透、能自己操控部署细节，我建议直接从 Apache Flink CDC 入手。

Flink CDC 是 Apache Flink 官方提供的变更数据捕获能力，底层封装了 Debezium，内置了 MySQL CDC、PostgreSQL CDC、Oracle CDC、SQL Server CDC 等连接器，用户只要写 Flink SQL 就能把数据库的增删改记录捕获成流式数据。这篇教程面向的是 Flink standalone 模式，也就是独立集群部署，不依赖 K8s 或 YARN，最直接、最无脑，也最适合先把原理和操作链路跑通。无论你之前有没有接触过 Flink，只要跟着一步步操作，都能把 MySQL CDC 和 PostgreSQL CDC 部署出来，并实现真实的增量同步。

我先说清楚这篇内容能解决什么问题：怎么下载安装 Flink standalone 集群、怎么准备 MySQL 和 PostgreSQL 的 CDC 源端环境、连接器 JAR 包应该扔到哪里、Flink SQL 要怎么写、作业怎么启动、怎么把变更记录接入 Kafka 消息队列，以及最容易被卡住的时区、主键、server-id 冲突这几个典型排障场景。如果你想在本地环境快速验证一套可用的 CDC 链路，或者准备给后端同事出一份落地的同步方案，都可以直接参考。

1. 整体设计与部署模式选型

1.1 CDC 的本质：为什么 Flink 能直接读到数据库变更

在动手部署之前，建议先把 CDC 的工作机制想清楚，不然很多问题出现了你根本不知道从哪查起。CDC 全称是 Change Data Capture，核心思路是让外部系统像消费者一样读取数据库的事务日志。MySQL 的 binlog、PostgreSQL 的 WAL（Write-Ahead Log）都是数据库底层天然记录的变更历史，Flink CDC 做的事就是把日志解析成结构化的变更事件流，然后交给下游算子处理。

具体到 Flink CDC 的架构，它内部集成了 Debezium 引擎，Debezium 连接数据库、读取日志、解析成 MySql Connector Records 格式，Flink CDC 再把它转换成 Flink 的 RowData 数据格式。所以整个链路的瓶颈通常不在这层框架，而在数据库日志的保留策略和网络带宽。

用 Flink SQL 写 CDC 作业有个非常大的好处：你不用关心底层 record 转换，也不需要写 Java 代码，要做的只是建 Source 表、建 Sink 表、写一条 INSERT INTO SELECT 语句，Flink 的 SQL Planner 会自动生成完整的流式计算任务。这非常适合快速开发，也方便非 Java 背景的数据同学直接上手。

1.2 standalone 模式与其它模式的取舍

Flink 集群部署有很多种方式：standalone、YARN、K8s、Native Kubernetes。这里我坚持推荐 standalone，原因是它足够轻。下载二进制包解压之后无非就是改两个配置文件，然后执行 start-cluster.sh，一台机器上就能把 JobManager 和 TaskManager 都跑起来。

standalone 模式适合的场景很清晰：学习阶段、功能验证、小规模数据同步、内网测试环境。它的缺点也很明显：缺少资源隔离，JobManager 单点没有自动 HA，生产环境大规模跑任务时建议还是走 K8s 或 YARN，让 Flink 之外的调度平台去管理资源和故障恢复。但对于这篇教程的目标——把 MySQL CDC 和 PostgreSQL CDC 完整跑通，standalone 是最合适的，它把变量降到最少，方便排查问题。

我见过不少人一上来就拿生产环境的 K8s 配置折腾，结果很多时候连“JAR 包没放进 lib 目录”这种基础问题都要排查半天。建议先 standalone 跑通，再谈容器化，这是我认为最稳的学习路径。

1.3 整体部署拓扑与目标链路

在这篇教程里，我们搭建一条完整的实时同步链路：

• Flink standalone 集群（1 个 JobManager + 1 个 TaskManager）
• 源端 MySQL 8.0 实例，开启 binlog
• 源端 PostgreSQL 13 实例，开启 logical replication
• 依赖连接器：flink-cdc-pipeline-connector-mysql、flink-cdc-pipeline-connector-postgres 或 SQL 连接器 JAR
• 可选下游：Kafka topic 或 Flink 控制台输出

最终我们希望达成的效果：在 MySQL 或 PostgreSQL 里执行 INSERT / UPDATE / DELETE，Flink 作业能实时捕获这些变更，并输出到日志或消息队列。这就是 CDC 链路最核心的能力。

2. 环境准备与版本选型

2.1 Flink 与连接器版本对应关系

版本问题是 CDC 部署的第一道坑。我在实际部署中踩过因为 Flink 1.14 和 CDC 2.x 不兼容导致作业反复重启的坑，所以这里建议直接用较新的稳定搭配：Flink 1.18.x + Flink CDC 3.1.x。这套组合比较稳定，文档也全。

为什么不建议用太老的 Flink 1.13 搭配 CDC 2.2？因为老版本有两个问题：一是 MySQL CDC 2.2 不支持 Flink 1.14+ 的某些新特性，二是 checkpoint 和状态恢复机制的兼容性不够好。如果你因为公司环境原因必须用老版本，建议把官方文档的版本矩阵截图存下来逐项核对。

版本选型可以按下面这张表来对照：

2.2 JDK 环境与 Flink 二进制包准备

Flink 1.18 依赖 JDK 8 或 JDK 11，我测试用的是 JDK 1.8 和 OpenJDK 11 都跑过，都没有问题。JDK 17 虽然新版 Flink 有支持，但建议别冒险，有些连接器和 JDK 17 的模块访问限制会导致 DataStream 作业抛异常。

下载地址就直接去 Apache Flink 官网找 flink-1.18.1-bin-scala2.12.tgz，解压到 /opt/flink 这种路径：

bash复制tar -zxvf flink-1.18.1-bin-scala2.12.tgz -C /opt
mv /opt/flink-1.18.1 /opt/flink

然后确认 JAVA_HOME 已配置：

bash复制echo $JAVA_HOME
/usr/lib/jvm/java-8-openjdk-amd64

启动 standalone 集群前，先检查一下配置 conf/flink-conf.yaml。单机验证场景下，我习惯把 TaskManager 内存调大一点，避免大表同步的时候 OOM：

yaml复制jobmanager.memory.process.size: 2048m
taskmanager.memory.process.size: 4096m
taskmanager.numberOfTaskSlots: 4
parallelism.default: 1

2.3 连接器 JAR 包获取方式

Flink CDC 与传统 Flink connector 有一个区别：它有两种使用形态。一种是基于 Flink SQL 直接使用，需要把 flink-sql-connector-mysql-cdc-3.1.0.jar 放到 Flink 的 lib 目录；另一种是借助 YAML 配置文件的 Pipeline 模式，需要下载对应 pipeline connector。

这里我强烈建议新手使用 Flink SQL 形态，因为你只需要下载这俩文件：

• flink-sql-connector-mysql-cdc-3.1.0.jar
• flink-sql-connector-postgres-cdc-3.1.0.jar

把这两个 JAR 包扔到 Flink 的 lib 目录之后重启 Flink 集群：

bash复制cp flink-sql-connector-*.jar /opt/flink/lib/
/opt/flink/bin/stop-cluster.sh
/opt/flink/bin/start-cluster.sh

提示：JAR 包版本必须与 CDC 版本完全一致，否则 SQL 客户端会直接报 ClassNotFoundException。另外连接器 JAR 包不要用瘦包，必须用 shaded 包，否则会把缺失的依赖类带出来。

3. MySQL CDC 部署实操

3.1 MySQL 源端配置：binlog 必须开启

MySQL 侧如果没开 binlog，Flink CDC 作业会卡在初始化阶段，报错信息通常是 Could not find first log file name in binary log index file 或 The server is not configured to use binlog。

先看当前 MySQL 的 binlog 状态：

sql复制SHOW VARIABLES LIKE 'log_bin';
SHOW VARIABLES LIKE 'binlog_format';
SHOW VARIABLES LIKE 'binlog_row_image';

如果 log_bin 是 OFF，就需要修改 MySQL 配置文件 /etc/my.cnf：

ini复制[mysqld]
server-id=223344
log_bin=mysql-bin
binlog_format=ROW
binlog_row_image=FULL
expire_logs_days=7
max_binlog_size=128M

配置完成之后重启 MySQL：systemctl restart mysqld 或 service mysql restart。然后验证配置生效：

sql复制SHOW MASTER STATUS;
+------------------+----------+--------------+------------------+-------------------+
| File             | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set  |
+------------------+----------+--------------+------------------+-------------------+
| mysql-bin.000003 |      157 |              |                  |                   |
+------------------+----------+--------------+------------------+-------------------+

看到有 binlog 文件名和支持点位置，源端就 OK 了。

有个细节需要特别留意：expire_logs_days 这个参数建议设置成 7 天以上。Flink 作业长时间暂停或者做全量快照恢复时，如果 binlog 已经被 MySQL 清理，作业重启就找不到位点，只能重新做全量同步。我就吃过这个亏，测试环境 binlog 只保留 12 小时，结果周末任务停了 2 天，周一恢复就报 FileNotFound。

3.2 创建 CDC 账号并赋权

CDC 读取 binlog 需要专用账号，不推荐直接用 root。创建账号时注意，要给 SELECT、RELOAD、SHOW DATABASES、REPLICATION SLAVE、REPLICATION CLIENT 权限：

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

然后单独给业务库加 SELECT 权限，比如我们要同步 demo 库：

sql复制GRANT SELECT ON demo.* TO 'flink_cdc'@'%';
FLUSH PRIVILEGES;

MySQL 8.0 的账号认证插件默认是 caching_sha2_password，有些客户端连不上，建议确认一下：

sql复制SELECT user, host, plugin FROM mysql.user WHERE user='flink_cdc';

如果是 caching_sha2_password 且出现认证失败，改成 mysql_native_password：

sql复制ALTER USER 'flink_cdc'@'%' IDENTIFIED WITH mysql_native_password BY 'flink_cdc_pass';

这一步在 MySQL 8.0 + 某些老版本连接器组合下非常关键，不做的话连接阶段直接抛 Public Key Retrieval is not allowed。

3.3 编写 Flink SQL 并启动作业

启动 Flink 集群后，进入 SQL 客户端：

bash复制/opt/flink/bin/sql-client.sh embedded

在 SQL 客户端里执行建表语句。MySQL CDC Source 表定义如下：

sql复制CREATE TABLE mysql_users (
  id INT,
  name STRING,
  email STRING,
  update_time TIMESTAMP(3),
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = '192.168.1.101',
  'port' = '3306',
  'username' = 'flink_cdc',
  'password' = 'flink_cdc_pass',
  'database-name' = 'demo',
  'table-name' = 'users',
  'server-id' = '5400-5404',
  'scan.incremental.snapshot.enabled' = 'true',
  'debezium.snapshot.mode' = 'initial',
  'debezium.time.precision.mode' = 'adaptive_time'
);

primary key 是必须的，Flink CDC 要靠它做 Changelog 的 upsert 语义，没有主键的表虽然也能读，但无法正确处理重复数据和删除事件，所以这里必须显式声明。

server-id 的范围非常重要。单独一个作业时你可以写一个固定值，但多作业读同一个 MySQL 实例时，如果 server-id 重复，MySQL 会误认为客户端异常断线重连，会导致其中一个任务一直报 Binlog is not configured correctly 或连接被 kill。规范做法是：每个作业分配一段范围，例如 5400-5404。

我们再创建 Kafka Sink 表把变更数据发出去：

sql复制CREATE TABLE kafka_users_sink (
  id INT,
  name STRING,
  email STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'topic' = 'ods_users',
  'properties.bootstrap.servers' = '192.168.1.102:9092',
  'properties.group.id' = 'flink-cdc-group',
  'format' = 'debezium-json',
  'sink.partitioner' = 'default'
);

然后执行数据接入语句：

sql复制INSERT INTO kafka_users_sink
SELECT id, name, email FROM mysql_users;

作业提交后，SQL 客户端界面会显示 SUBMITTED，正常情况会看到作业状态变更为 RUNNING。往 MySQL 里插一条数据，观察 Flink 日志或者 Kafka 消费者，就能看到变更记录。

3.4 MySQL CDC 全量快照与增量衔接原理

有一个重要的机制值得展开讲：Flink MySQL CDC 默认的 initial 模式启动时会先做全量快照，再无缝切换到 binlog 增量读取。这个全量阶段不是用 SELECT 查出全表再读，而是使用一致性快照，参考点取自 binlog 的位点，这样全量和增量之间不会丢数据。

在实现上，新版连接器基于增量快照框架把表分成 chunk 并行读取，并且用 checkpoint 记录进度。如果你的表非常大，比如几亿行，可以通过 scan.incremental.snapshot.chunk.size 调整 chunk 大小，默认是 8096 行，这个值调的太小会导致任务频繁请求数据库，调太大则占用更多内存。经验值是 1 万到 2 万行，具体视单行平均大小而定。

全量跑的时候，注意连接器会对读取的表加一个短暂的锁来获取一致性位点，所以如果业务高峰期要启动同步，最好提前评估会不会影响业务。其实新版连接器对无主键大表的锁策略做了优化，但最稳妥的做法还是业务低峰期启动初始同步。

4. PostgreSQL CDC 部署实操

4.1 主要配置 wal_level 与复制账号

PostgreSQL CDC 走的是逻辑复制能力，基于 WAL 日志解码成逻辑变更流。和 MySQL 类似，也需要改配置。修改 postgresql.conf：

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

其中 wal_level = logical 是最核心的，它会在 WAL 中额外写入逻辑解码所需的信息。改完需要重启 PostgreSQL：

bash复制systemctl restart postgresql

然后创建复制专用账号：

sql复制CREATE USER flink_cdc WITH PASSWORD 'flink_cdc_pass';
ALTER USER flink_cdc WITH REPLICATION;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO flink_cdc;
GRANT USAGE ON SCHEMA public TO flink_cdc;

逻辑复制槽需要什么样的权限？如果你想更精细，还可以给 pg_create_subscription 权限，但 Flink CDC 的连接器一般会用现成的 replication slot 工具创建槽位，所以上面这些就够了。

4.2 创建发布（Publication）

PostgreSQL 逻辑复制中，发布（Publication）定义了哪些表可以被订阅。Flink CDC 连接器默认会自动创建发布，名字一般是 flink_cdc_publication，但为了让权限可控，我更建议提前手动创建：

sql复制CREATE PUBLICATION dbz_publication FOR TABLE users;

如果你希望某个 schema 的所有表都能同步：

sql复制CREATE PUBLICATION dbz_publication FOR ALL TABLES IN SCHEMA public;

发布没建好，Flink 作业启动时会报 relation "public.users" does not exist 或者 publication "dbz_publication" does not exist。所以这个步骤最好手动提前做，避免权限不足无法自动创建。

4.3 编写 PostgreSQL CDC 的 Flink SQL

PostgreSQL CDC 的 Source 表写法：

sql复制CREATE TABLE pg_orders (
  id INT,
  user_id INT,
  amount DECIMAL(10, 2),
  created_at TIMESTAMP(3),
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '192.168.1.103',
  'port' = '5432',
  'username' = 'flink_cdc',
  'password' = 'flink_cdc_pass',
  'database-name' = 'postgres',
  'schema-name' = 'public',
  'table-name' = 'orders',
  'slot.name' = 'flink_slot_01',
  'decoding.plugin.name' = 'decoderbufs',
  'debezium.snapshot.mode' = 'initial',
  'debezium.plugin.name' = 'decoderbufs'
);

注意 slot.name 是复制槽的名字，Flink 作业会基于它进行 WAL 位点记录。一个复制槽同一时间只能被一个消费端使用，所以如果任务重启后一直提示 replication slot is active，说明旧的连接没有释放。

decoding.plugin.name 一般用 decoderbufs，如果使用 wal2json 或者其他插件则需要对应调整。如果你是第一次测试，我建议直接用默认稳定组合 decoderbufs，因为 wal2json 在处理整数类型时输出格式和 decoderbufs 有差异，有些数据类型映射会有出入。

再建一个控制台打印表或者 Kafka Sink，然后执行接入：

sql复制INSERT INTO kafka_orders_sink
SELECT id, user_id, amount, created_at FROM pg_orders;

4.4 PostgreSQL 类型映射与时间字段处理

PostgreSQL 的类型体系比 MySQL 更丰富，所以做 CDC 时特别容易遇到类型映射问题。比如 numeric 类型映射到 Flink 的 DECIMAL 时，如果精度没有被正确声明，下游写 Kafka 的 Debezium JSON 会带出非常多位数，导致消费者解析困难。

我的建议是在建表 DDL 里显式声明精度：

sql复制amount DECIMAL(10, 2)

而不是直接写 amount DECIMAL。

PostgreSQL 的 timestamp with time zone 类型会被映射成 TIMESTAMP_LTZ，如果你需要保留时区语义，请在 Flink SQL 中也使用 TIMESTAMP(3) WITH LOCAL TIME ZONE。但下游如果不是很依赖时区信息，直接用 TIMESTAMP(3) 也能跑，只是显示的时间会转成 Flink 进程的时区。

生产环境我推荐统一约定：所有时间字段按 UTC 存储，Flink 作业统一使用 Asia/Shanghai 时区做展示转换，避免下游计算出现 8 小时偏差。具体在 Flink 配置里加：

yaml复制env.java.opts: "-Duser.timezone=Asia/Shanghai"

或者在 SQL 客户端初始化脚本执行 SET 'sql-client.execution.result-mode' = 'tableau'; 时配合 TIME ZONE 'Asia/Shanghai' 语义。

5. 常见问题与排查技巧实录

5.1 作业启动阶段的问题

问题一：ClassNotFoundException / NoClassDefFoundError

这个 90% 的原因是连接器 JAR 没有放进 lib 目录，或者放进去了没有重启集群。还有一种可能是你把连接器放到了 Flink 的 plugins 目录，而不是 lib 目录。检查方法：

bash复制ls /opt/flink/lib/ | grep cdc

如果能找到 JAR，至少会输出 flink-sql-connector-mysql-cdc-3.1.0.jar，没有则确认 cp 命令的路径。

问题二：Could not find first log file name in binary log index file

这个报错包含两种情况。第一种是 MySQL 没开 binlog，按 3.1 节方法解决；第二种是连接器拿着 binlog 位点去找文件，结果日志文件被清理了。如果你的 expire_logs_days 设置的太小，调大之后检查 SHOW BINARY LOGS; 确认 binlog 文件确实存在。

问题三：连接超时但 MySQL 本身空闲

如果你使用固定 server-id 而不是区间，且存在多作业并发的情况，MySQL 会强行断开连接。解决办法是把 server-id 配置成 5400-5404 这种区间形式，这样并发 source 实例读取时不会互相冲突。

5.2 运行过程中的问题

问题四：PostgreSQL replication slot 一直 active 导致重启失败

这种情况多数是 Flink 作业没有正常停止，状态后端和复制槽没有释放。可以登录 PostgreSQL 手动查看：

sql复制SELECT * FROM pg_replication_slots;

如果确认 flink_slot_01 已经没有实际消费端，执行：

sql复制SELECT pg_drop_replication_slot('flink_slot_01');

注意正式环境不要随便删 slot，因为那个位点直接关系到下游从哪继续消费。

问题五：数据延迟越来越大

观察 Flink UI 的反压水位，很可能出现两种情况：一是 MySQL 源表没有主键导致全量阶段一直在扫描；二是 Sink 端 Kafka 分区数不足，写入出现热点。如果是前者，给源表加主键或调整 Start from earliest 策略；后者就扩大 Kafka topic 分区数并调整 sink.partitioner。

问题六：时间字段差 8 小时

Flink CDC 会把时间以 UTC 写入 Kafka Debezium JSON。如果你的下游直接展示原始值，就会看到差 8 小时。解决方案是统一 Flink 进程时区，或者在 Sink 表里用 LOCALTIME 转换。个人建议前者，全局层面一次解决。

问题七：缺少主键导致无法更新删除记录

Flink SQL 在 changelog 模式下如果没有主键，默认用 append-only 流，DELETE 事件无法进入下游主键模型。建表时如果源表确实没主键，可以考虑加一个基于多列的主键声明，只要那些列能联合唯一就行。

5.3 常用排查指令速查

5.4 一点避坑心得

我实际跑这些作业踩过最多的坑其实不是框架本身，而是“环境不一致”。比如本机 MySQL 是 5.7，测试环境是 8.0，连接器在解析时间类型和 JSON 类型时行为有细微差异；再比如 PostgreSQL 的纳秒精度在 Debezium 输出里带了很多位小数。

所以建议搭建一套 Docker Compose 的本地环境，把 Flink、MySQL、PostgreSQL、Kafka 四个容器固定版本跑起来。这样不论你怎么折腾，环境始终一致，排查问题效率会高非常多。用 Docker 跑 Flink standalone 其实也简单，映射 8081 和 6123 端口，挂载 lib 目录即可。

另外关于 checkpoint 配置，CDC 作业强烈建议打开 checkpoint。每 10 秒自动做一次 checkpoint，源端位点也会存进去，出问题重启时直接从上次 checkpoint 恢复，不会重复消费大量数据。

yaml复制execution.checkpointing.interval: 10s
state.backend: rocksdb
state.checkpoints.dir: file:///opt/flink/checkpoints

写在最后的小提示

如果你是自己练习，建议第一次跑通后把作业停掉，然后在 MySQL 和 PostgreSQL 里同时执行几条 UPDATE 和 DELETE，再重启 Flink 作业，观察它能不能从 checkpoint 恢复并继续同步。这个过程能帮你真正理解 CDC 作业的状态恢复机制。

我个人在实际操作中的体会是：Flink CDC 的难点从来不在于“跑起来”，而在于把源端数据库的日志配置、账号权限、类型映射、以及异常恢复路径都理解透。这篇提到的每一步，都是我从反复重启作业和凌晨三点翻日志的经历里沉淀出来的，照着操作可以省掉不少弯路。数据库版本和连接器版本如果跟你手头的环境不一样，第一件事永远是查版本兼容矩阵，别硬试。