正值某支付平台凌晨大促订单表像瀑布一样往下掉数据。你以为是流量分析平台那边延迟结果查下来发现他们读的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工具做T1同步的方案这条链路最大的优势是全量阶段只需要读一遍源库存量数据不会重复抽取增量阶段延迟控制在秒级万一整个链路崩了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里才包含完整的镜像数据。如果设置成MINIMALFlink 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.DriverMySQL用同一个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.0PostgreSQL JDBC驱动42.x用于连GreenPlumGreenPlum是PostgreSQL 8.x基础上改的6.x和7.x的GreenPlum都能用42.x驱动连。依赖放在Flink的lib目录下即可。如果用的是Flink SQL客户端把下面几个jar包放进去flink-sql-connector-mysql-cdc-2.3.0.jarflink-connector-jdbc_2.12-1.14.x.jarmysql-connector-java-8.0.x.jarpostgresql-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.modeinitial表示作业第一次启动时先全量读一遍再自动衔接增量。如果改成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_timeCREATE 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 SQLCREATE 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的metricsAlertmanager发告警到群。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的语义上框架保障不丢但重复靠业务逻辑去兜底才是真正万无一失。