正值某支付平台凌晨大促，订单表像瀑布一样往下掉数据。你以为是流量分析平台那边延迟，结果查下来发现他们读的MySQL从库已经落后主库快十分钟，而业务方又催着要把订单明细实时推到GreenPlum做报表、还要同步一份到另一个MySQL供下游服务查询。这种场景相信不少做数据开发的兄弟都撞见过：一边是源端数据库不能停、不能锁表，一边是下游要求秒级延迟，传统的离线抽数完全顶不住。那段时间我接了个私活，专门做Flink实时读取MySQL增量日志数据并写入GreenPlum/Mysql的同步链路，效果还算理想，抽空把整个项目的踩坑过程整理出来，希望对做实时数仓和数据同步的朋友有点帮助。
这个项目解决的核心问题其实就一件事：在不影响源库正常读写的前提下，把MySQL里的行级变更毫秒级抓到流里，经过必要的清洗和补全，再双写到GreenPlum和另一套MySQL。适合正在搭实时数仓、又不想引入太重框架的同学参考。下面我就从方案选型、核心配置、写入实现到生产问题排障，把这条链路从头到尾拆开讲。
1. 项目背景与整体方案选型
1.1 为什么非要用Flink做增量同步
最初接到需求时，我的第一反应是能不能用现成的同步中间件。市面上做MySQL同步的工具不少，开源的有基于Python的mysql-binlog-connector、基于Java的canal，以及商业化的各种数据同步服务。但挨个试下来都有点不对味：
- 基于单线程解析binlog的轻量工具，扩展性差，吞吐一上来就撑不住；
- canal本身很强，能订阅binlog并转成消息，但链路里还得自己搭消息队列、自己写消费端，Flink引入后相当于把解析、处理、汇出三层揉在一起做；
- 直接用消息队列的CDC功能，又会受限于消息体大小和消费端逻辑，处理脏数据的能力很弱。
而Flink体系里的Flink CDC连接器，底层基于Debezium，天然把binlog解析这一层封装好了。我只需要在Flink SQL里建一张映射表，就能像读普通流式表一样消费MySQL的INSERT/UPDATE/DELETE事件。再加上Flink的Checkpoint机制能保证精确一次语义，同一套作业里可以同时挂上GreenPlum和MySQL两个sink，两个目标端的数据一致性天然统一。这个“一套流，双端写”的能力，是之前分开搭工具很难做到的。
1.2 CDC链路的核心组成架构
整个链路的组成并不复杂，核心是三层：
源端：MySQL实例开启binlog且采用ROW格式，业务表保持主键或唯一键可用。应用侧业务完全无感知，不需要停读写。
中间层：Flink作业启动后，第一步做全量快照。Flink CDC连接器会自动先读一次全量数据，再无缝切换到增量binlog事件，这个切换过程对外表现为一次“无中断的初始化”，对下游来说就像数据从零开始连续流动。
目标端：GreenPlum接收的是贴源明细数据，用于后续的报表汇总；MySQL接收的是下游微服务要用的副本表。两个sink通过JDBC批量写入，各配各的连接池和批次参数，互不影响。
对比用离线ETL工具做T+1同步的方案，这条链路最大的优势是：全量阶段只需要读一遍源库，存量数据不会重复抽取；增量阶段延迟控制在秒级；万一整个链路崩了，Checkpoint恢复后binlog可以从上次记录的位点继续消费，不做重复写入。
2. 基础设施与依赖准备
2.1 源端MySQL的binlog配置检查
动手前先把源端数据库的binlog相关参数确认到位，这是最容易翻车的地方。我平时排查的顺序是：
show variables like 'log_bin'; show variables like 'binlog_format'; show variables like 'binlog_row_image'; show variables like 'server_id';必须要保证的几项是：
- log_bin必须为ON，否则没有增量日志可读；
- binlog_format必须为ROW，因为只有ROW模式才能在日志里记录每一行的字段变更前后值；
- binlog_row_image建议设为FULL，这样binlog里才包含完整的镜像数据。如果设置成MINIMAL，Flink CDC解析时虽然能拿到主键，但非主键字段的值是拿不到的，这在需要同步完整字段时会踩大坑；
- server_id不能在多个同步任务里重复。Flink CDC连接器本质上是模拟一个MySQL从库去拉binlog，每个连接器实例都要用独立的server_id，不然会和真实的从库以及其他任务相互踢掉线。
特别提醒一句，修改binlog_format之后，旧的binlog文件里可能还有非ROW格式的事件，不过对增量同步没有影响。为了干净起见，改动完可以flush logs切一个新文件，让Flink从新文件开始消费。
2.2 GreenPlum与MySQL作为目标端的差异
目标端一个走分析、一个走业务，我对两个sink的选型考量是不一样的。
GreenPlum这边，它是基于PostgreSQL生态的分布式数仓，写入时最忌讳高频小事务。理想的方式是把数据攒批，单次提交尽量大。同时GreenPlum的primary key约束本身不是强约束，需要靠数据管道的幂等逻辑来保证不重复。我的做法是：sink端用主键做upsert，并且在建表时指定随机分布键或者按业务键分布，避免全表重分布。
MySQL那边则相反，它需要控制单个事务的大小，避免大事务带来锁竞争和主从延迟。我设置批量大小在几百条级别，flush间隔小于1秒，既不至于太大，也不会因为频繁提交刷爆日志。
两边的连接器实现也不同。GreenPlum我用Flink JDBC连接器配合org.postgresql.Driver，MySQL用同一个JDBC连接器配合mysql驱动，两张输出表在Flink SQL层面看起来差别不大，但实际运行时的提交策略需要针对性调参。
2.3 版本匹配与依赖清单
版本选型这部分我经历过不少折腾，总结下来可以少走弯路的组合是：
- Flink 1.14及以上，因为我用的是Flink CDC 2.x系列，2.x对增量快照、动态加表支持得更完善；
- Flink CDC 2.3+，这个版本里mysql-cdc连接器的类名和配置项已经稳定；
- MySQL驱动5.1.49或8.x，取决于你的MySQL是5.7还是8.0；
- PostgreSQL JDBC驱动42.x，用于连GreenPlum，GreenPlum是PostgreSQL 8.x基础上改的，6.x和7.x的GreenPlum都能用42.x驱动连。
依赖放在Flink的lib目录下即可。如果用的是Flink SQL客户端，把下面几个jar包放进去：
- flink-sql-connector-mysql-cdc-2.3.0.jar
- flink-connector-jdbc_2.12-1.14.x.jar
- mysql-connector-java-8.0.x.jar
- postgresql-42.x.jar
这里有个容易踩的坑：Flink CDC连接器自带Debezium依赖，如果写在pom里时provided和compile范围没处理好，很容易出现jar冲突，报错通常是NoSuchMethodError或者ClassNotFoundException。建议在Flink提供的环境里跑就用provided，产品环境直接把连接器jar扔到lib目录。
3. 核心实现：Flink CDC采集与处理
3.1 源表CDC连接器的Flink SQL定义
这里我以业务中常见的orders订单表为例，结构大概是id、order_no、user_id、amount、status、create_time。在Flink SQL中定义源表如下：
CREATE TABLE orders_src ( id BIGINT NOT NULL, order_no STRING, user_id BIGINT, amount DECIMAL(10, 2), status INT, create_time TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = '192.168.1.101', 'port' = '3306', 'username' = 'bidata', 'password' = 'xxxxxx', 'database-name' = 'trade', 'table-name' = 'orders', 'scan.incremental.snapshot.enabled' = 'true', 'server-id' = '5400-5404', 'scan.startup.mode' = 'initial' );几个关键配置我展开解释一下：
scan.incremental.snapshot.enabled这个参数决定Flink CDC是全表加锁扫一遍还是分段chunk扫描。老版本默认会用全局锁去拿位点，锁表期间源库写入会被阻塞。这个参数设为true之后，会按主键拆成多个chunk并行扫描，配合binlog位点做切分，几乎不影响线上业务。
server-id写成5400-5404，是一个范围而不是单个值，这个是关键。因为并行度超过1时，每个并发的读取子任务都需要一个独立server_id，如果我这里写死一个，并行度一起来就会撞id。
scan.startup.mode=initial表示作业第一次启动时，先全量读一遍，再自动衔接增量。如果改成latest-offset，则只从当前时间点消费增量，存量数据不管，适合目标表已经有过一轮数据的情况。
跑起来之后，我想快速验证数据有没有抓到，可以先不接真正的sink，而是临时建一个print表观察输出：
CREATE TEMPORARY TABLE print_sink ( id BIGINT, order_no STRING, user_id BIGINT, amount DECIMAL(10, 2), status INT ) WITH ('connector' = 'print'); INSERT INTO print_sink SELECT id, order_no, user_id, amount, status FROM orders_src;如果print表里能看到数据的增删改变化，说明CDC通道是通的，可以继续往下接sink。
3.2 数据清洗与业务字段补全
真正接业务的时候，源数据往往不能直接往下游扔。我在这条链路上做了三件常见的事：
第一件是给数据增加一个_process_time字段，统一记录Flink处理这条数据的时间。下游如果需要区分“业务发生时间”和“数据进入数仓时间”，这个字段就是关键依据。
第二件是把一些枚举值做翻译。比如status在源库是数字，1代表待支付、2代表已支付、3代表已发货，直接同步到数仓里下游还得查字典。我在Flink SQL里用CASE WHEN做了一层翻译，把status转成字符串带出去。
第三件是过滤掉非法数据。有些历史脏数据，比如amount为负或者user_id为空，不该进数仓。我在中间层加一个filter过滤掉这些行，顺便把这个筛选逻辑埋一个计数器，方便Metrics监控里查询。
一个典型的中间层视图长这样：
CREATE VIEW orders_cleaned AS SELECT id, order_no, user_id, amount, CASE status WHEN 1 THEN 'PENDING' WHEN 2 THEN 'PAID' WHEN 3 THEN 'SHIPPED' ELSE 'UNKNOWN' END AS status_text, COALESCE(create_time, CURRENT_TIMESTAMP) AS business_time, CURRENT_TIMESTAMP AS process_time FROM orders_src WHERE user_id IS NOT NULL AND amount > 0;3.3 动态追加开发中的Schema变更处理
生产环境里，业务方改表结构是常事。Flink CDC 2.x对加了新列的情况处理得比较平滑：源端ALTER TABLE加列之后，新数据里的新列会以null或默认值出现在目标端，前提是Flink SQL里对应的字段已经定义好。
如果新加的列是必填的，并且下游要立即消费，就得停作业、更新SQL、从savepoint恢复。这个操作本身不算复杂，但要注意：恢复的时候从savepoint拉起，binlog位点会接着上次的位置走，千万不能用无savepoint的restart，否则会丢位置。
我自己的经验是，能不依赖自动加列就不要依赖，最好在提交作业之前把上下游的表结构对齐一次，之后再变更走正式流程。这样排障的复杂度能低很多。
4. 写入GreenPlum与MySQL的实现细节
4.1 GreenPlum侧输出表与JDBC配置
先看GreenPlum的输出表定义。我这里建了一张目标表，结构和源表基本一致，但额外加了process_time和sync_time：
CREATE TABLE gp_orders ( id BIGINT, order_no VARCHAR(64), user_id BIGINT, amount DECIMAL(10, 2), status_text VARCHAR(32), business_time TIMESTAMP, process_time TIMESTAMP ) DISTRIBUTED BY (id);GreenPlum的建表语句和标准PostgreSQL有点区别，重点是DISTRIBUTED BY这行，它决定了数据在哪个Segment节点上存储。我用id做分布键，后续按订单维度做join时能避免节点间shuffle。
Flink SQL里连接GreenPlum的方式如下：
CREATE TABLE gp_sink ( id BIGINT, order_no STRING, user_id BIGINT, amount DECIMAL(10, 2), status_text STRING, business_time TIMESTAMP(3), process_time TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:postgresql://192.168.1.120:5432/analytics', 'table-name' = 'gp_orders', 'username' = 'gpadmin', 'password' = 'xxxxxx', 'sink.buffer-flush.max-rows' = '1000', 'sink.buffer-flush.interval' = '2s', 'sink.max-retries' = '3' );JDBC连接器的sink默认不是严格的upsert，如果上游带来重复主键，会产生重复行。判断依据是：声明了PRIMARY KEY之后，JDBC连接器会尝试执行INSERT ON DUPLICATE KEY UPDATE。但这个语法在GreenPlum里是不支持的，GreenPlum支持的是INSERT ON CONFLICT。
所以在对GreenPlum做upsert时，我换了另一种思路：不改连接器，而是在主键冲突处理上用“先删后插”的方式。做法是在目标表上建立唯一索引，然后通过自定义sink或前置debezium配置，把DELETE事件和INSERT事件按主键对齐。如果只是普通追加，不要求幂等，直接用INSERT就行。
这里实测下来，最省心的方案是：GreenPlum版本如果支持，用支持ON CONFLICT的PostgreSQL方言；如果不支持，就在目标表上定期做merge。我的项目中GreenPlum是7.x版本，通过修改JDBC连接器的方言配置或者自己写一个sink函数，才能顺畅实现幂等写入。如果写不了自定义sink，建议在目标表上不要设置主键约束，用定时清理加去重查询的方式兜底。
4.2 MySQL侧输出表与主键幂等写入
回到另一个MySQL实例，我这里建同样的orders表，但加了一个全局唯一的synced_id字段作为幂等键，避免重复跑任务时把数据插重：
CREATE TABLE mysql_orders_sink ( id BIGINT NOT NULL, order_no VARCHAR(64), user_id BIGINT, amount DECIMAL(10, 2), status_text VARCHAR(32), business_time DATETIME, process_time DATETIME, synced_id BIGINT, PRIMARY KEY (id), UNIQUE KEY uk_sync (synced_id) );对应的Flink SQL：
CREATE TABLE mysql_sink ( id BIGINT, order_no STRING, user_id BIGINT, amount DECIMAL(10, 2), status_text STRING, business_time TIMESTAMP(3), process_time TIMESTAMP(3), synced_id BIGINT, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://192.168.1.150:3306/business_copy', 'table-name' = 'mysql_orders_sink', 'username' = 'sync_user', 'password' = 'xxxxxx', 'sink.buffer-flush.max-rows' = '500', 'sink.buffer-flush.interval' = '1s', 'sink.max-retries' = '3' );当Flink SQL表声明了PRIMARY KEY且目标是MySQL时，JDBC连接器会自动补成UPSERT语句，也就是INSERT INTO ... ON DUPLICATE KEY UPDATE。这样源表的主键和大字段一更新，目标表也会跟着更新，而不是简单插入重复行。这个行为在处理UPDATE事件时非常关键。
4.3 双sink共存与事务一致性保障
一个作业里同时多个sink时，Flink的Checkpoint机制是保障一致性的基石。具体一点说：Flink分布式快照会把CDC源表当前消费到的binlog位点和两个sink各自的写入状态一并存起来，之后无论哪个sink失败，只要从最近一次Checkpoint恢复，源端会从保存的位点重新读数据，两个sink也会回滚到这次Checkpoint对应的写入状态，最终效果是两端数据都不会丢也不会重。
不过要注意，Checkpoint只保证“事件至少处理一次”到“精确一次”的语义，取决于sink是否实现了事务性提交。JDBC sink默认是at-least-once，也就是说极端情况下可能重复写入。所以我的代码里在目标表上都做了幂等约束，让重复写入自然被拒掉。这是实时同步里特别实用的一套组合拳：框架保证不丢，业务表保证不重。
实际开发中，我会把两个sink放在一个INSERT INTO ... SELECT语句里同时执行：
INSERT INTO gp_sink SELECT id, order_no, user_id, amount, status_text, business_time, process_time FROM orders_cleaned; INSERT INTO mysql_sink SELECT id, order_no, user_id, amount, status_text, business_time, process_time, id AS synced_id FROM orders_cleaned;两个INSERT各自对应一个独立的sink任务，在同一个Flink作业里并行跑。只要各自sink的并行度合理，双写吞吐能同时打满，且不会互相拖慢。
5. 生产部署与性能调优
5.1 作业部署与资源分配建议
我一般使用Flink on YARN或Kubernetes部署，作业提交方式用application mode，这样提交后可以直接观察日志和Metrics。开发期间用session模式方便，但生产上session模式容易出现资源争抢，任务一多就互相踩。
资源分配方面，一个这样的双写链路，我给作业分配2个TaskManager，每个2个slot，并行度设置如下：
- 源端orders_src并行度设为2，让Flink CDC按主键chunk并行读全量；
- 中间算子并行度4，跑过滤和字段翻译；
- gp_sink并行度2，保持与GreenPlumSegment数量接近；
- mysql_sink并行度2。
这个配置在单表几千万数据、日增量百万行左右的规模下，能稳定跑到每秒几千条的吞吐，延迟按秒级。数据量再往上翻，优先加源端并行度和sink端并行度，Checkpoint间隔可以稍微调大一点。
5.2 关键性能参数调优记录
调优过程中我重点调过这几个参数，效果显著的都记下来了：
参数名作用推荐设置合理原因
execution.checkpointing.interval 控制Checkpoint触发频率 60s 间隔太短会导致频繁做快照，占用额外资源 execution.checkpointing.min-pause 控制两次Checkpoint最小间隔 30s 避免连续快照，给作业留出喘息 state.backend.type 状态后端类型 rocksdb 增量数据位点和去重状态都用RocksDB持久化，flink内存扛不住大状态 env.java.opts 增加JVM堆内存 -Xms2g -Xmx2g 源端chunk扫描和JDBC批处理都需要内存 sink.buffer-flush.max-rows 控制批大小 上面分别设了1000和500 两个目标端的行数阈值不同，GreenPlum可以更大
另外我还在作业里开了minibatch对中间聚合做了微调，但同步链路没有聚合窗口，其实收益不大，这里就不展开。
5.3 监控指标与告警配置
一条实时同步链路，最怕的就是没有健康监控。我主要监控四类指标：
- 源端binlog消费延迟：Flink CDC连接器暴露的current-fetch-timestamp与event时间戳差值，超过几十秒就告警；
- Checkpoint耗时和失败次数：连续失败2次以上看日志，多半是源库锁或网络不稳；
- 两个sink的写入延迟和失败重试数：JDBC连接器有metric能查到写入失败的重试次数；
- 目标端的表增长量：简单粗暴的对比查询，每小时对比源表行数和目标表行数，发现差异扩大就人工介入。
告警渠道我用的是生产环境已有的监控平台，Prometheus拉Flink的metrics，Alertmanager发告警到群。
6. 常见问题与故障排查实录
6.1 经典现场一：作业启动后一直卡在全量阶段
这是我第一次跑Flink CDC项目时遇到的坑。现象是作业运行了很久，source端没有增加数据，sink端没有任何写入。日志里能看到Splitting Table，但一直处于Splitting状态。
排查ping：全量快照阶段Flink CDC会拿SELECT * FROM orders去读取数据。如果业务方在源库上有一个长期运行的慢查询，或者有大表DDL锁，会阻塞快照查询。
解决方式：先看源库的processlist，杀掉阻塞的查询；如果表实在太大，不要等全量结束才接增量，把scan.incremental.snapshot.chunk.size调小，比如从8096改成2048，让chunk粒度更细，更快切换到增量消费。
还有一种可能性是server-id与其他同步任务冲突了，现象是源库binlog连接反复断开重连。这时候改一个全新的server-id范围，重启作业即可。
6.2 现场二：字段类型不一致导致写入失败
订单表里amount在源库是decimal(10,2)，目标表建表时不小心建成double，看起来能写进去，但GreenPlum端浮点精度会出问题，金额出现0.01的偏差。排查时看的是sink的异常日志：Batch entry ... cannot be cast。
解决方式：统一目标端字段类型为decimal(10,2)，并在Flink SQL里显式CAST，别让框架自动转。
另一个高频问题是，源表某字段为varchar，但值里混入了非法编码，尤其是emoji表情存进MySQL时如果字段不是utf8mb4会报错。此时目标表也必须是utf8mb4，否则写入会抛Incorrect string value。这个坑在双写时出现概率很高，建表时要格外注意字符集。
6.3 现场三：source与sink并行度不匹配导致的数据倾斜
某个现场我遇到GreenPlum一个Segment负载特别高，其他Segment很闲。后来发现是顺序id做分布键本身就有问题，因为id递增，同一时间产生的数据都落到同一个Segment上。
解决方式：把分布键从id改成order_no，也就是把业务键打散。同时把sink端并行度调成与Segment数一致，让每个Slot负责一个Segment的写入。
如果没法改分布键，还可以用随机分布，但那样后续join查询性能会比较差，所以还是优先选业务分布键。
6.4 一条很少人提到的实战经验：关于savepoint恢复
最后提醒一下，不管作业运行多久，都要保留最近的savepoint路径。同步链路一旦跑久了，binlog位点状态都攒在状态里，丢了重来就得全量再拉一遍。几千万的表现在重新拉全量，代价谁也受不了。
我自己的习惯是每天凌晨定时触发一次savepoint，至少保留最近7个。这样哪天发了新版本有问题，想回滚也方便。
7. 写在最后的体会
这个项目算是我做过的最典型的实时数据管道之一，表面上看只是把MySQL的binlog接到Flink、再写入两个目标端，实际上牵涉到binlog格式、快照与增量切换、幂等写入、监控运维等一系列细节。我做完之后最大的体会是：实时同步链路能不能稳定跑，有一半的功夫花在源端配置和目标端的建表合理性上，代码本身反而是最简单的那一层。建议第一次做这类项目的朋友，先把binlog相关配置和表结构对齐检查完，再开始写Flink SQL，能省下大量排障时间。另外，双写场景里一定要给目标表做好幂等键，别把可靠性完全押在Flink的语义上，框架保障不丢，但重复靠业务逻辑去兜底，才是真正万无一失。
