简介由尚硅谷研究院撰写的Flink CDC 3.0学习文档面向具备一定Flink基础的大数据处理研发人员系统讲解如何基于变更数据捕获技术实现MySQL到Doris的实时数据同步。内容首先界定CDC概念比较基于查询和基于Binlog两种方式的执行模式与延迟差异随后引出flink-cdc-connectors组件的读取能力并说明其适用的常见数据库。案例实操部分覆盖数据准备、DataStream与Flink SQL两种开发方式以及完整的Streaming ETL流程环境搭建、配置编写、任务启动和结果测试均提供清晰步骤同时强调开启Binlog、设置检查点等保证可靠性与数据一致性的关键设置。资料为单个docx文档大小145KB结构紧凑方便通读与随用随查。目前已有1788人浏览学习文档内附带可直接复用的代码示例和操作指南能够帮助读者独立完成Flink CDC项目部署并将MySQL到Doris的高效稳定同步方案迁移至实际业务场景中。1. CDC与Binlog为什么「实时同步」不是实时查询MySQLMySQL数据实时同步到Doris的Streaming ETL场景下Flink CDC 3.0常被误当成一个「实时查询MySQL」的工具。它的底层其实不靠查询连接器把自己模拟成一台MySQL从库订阅Binlog的变更事件流再交给Flink算子去加工。这个「监听日志」的思路决定了它为什么延迟低、为什么能捕获UPDATE和DELETE、为什么全量历史数据还要靠快照去补齐。搞懂CDC与Binlog的关系、启动策略和位点恢复机制才能判断一份同步配置为什么能work、什么时候会翻车。这份讲义把从MySQL到Doris的完整链路拆开讲了一遍值得按章节对照实际环境跑一遍。2. 基于查询与基于BinlogCDC的两条技术路线为什么后者才是流式底座一提到实时同步很多人的第一反应是「定时把数据查出来搬过去」。这个直觉在数据量小时没问题一旦表到了千万行任何周期的全量扫描都会变成主库的负担。CDC要解决的正是「如何以最低代价知道哪些数据变了」这个问题。围绕这个目标业界分化出两条路线基于查询和基于Binlog。两条路线的差别不只是技术选型它决定了你拿到的是一个批处理工具还是一条真正的流。2.1 基于查询的CDC为什么Sqoop、DataX只能做批基于查询的CDC核心思路是周期性发起SELECT用某个增量字段比如modify_time、自增id判断哪些行在两次查询之间发生了变化再把它们拉出来写入目标端。Sqoop、DataX是这条路线最常见的代表。它们的执行模式天然是Batch设定一个轮询间隔到点就全表扫描一次把超过上次水位线的记录带走。这里有一个绕不开的死角——删除。如果一行数据在主库被DELETE了它根本不会出现在下一次查询结果里目标端永远不知道这行已经消失。更麻烦的是查询压力。全表扫描的代价和数据量成正比轮询越频繁主库越吃力轮询越稀疏数据新鲜度越差。这个矛盾无法靠调优参数解决因为查询模型本身就不适合持续追踪变更。讲义里那张对比表把这种差距说得很直白执行模式是Batch、延迟高、会给数据库增加压力。数据量小的时候这些缺点还能忍一旦业务表到达亿级基于查询的同步就会变成主库的定时炸弹。2.2 基于Binlog的CDCCanal、Maxwell、Debezium的流式基因另一条路线是读Binlog。Binlog本身就是MySQL为了实现主从复制而设计的顺序日志主库把每一次INSERT、UPDATE、DELETE按提交顺序追加进日志从库拉取日志并回放。基于Binlog的CDC组件做的事情本质上是把自己伪装成一台MySQL从库订阅这份日志流。因为日志是顺序追加的天然就适合流式处理因为所有变更都在日志里DELETE也能被捕获不丢任何类型的数据因为它只是读日志不执行查询对主库几乎不增加压力。Canal、Maxwell、Debezium都是走这条路线的开源产品区别在于消费日志之后的封装方式。Flink社区的flink-cdc-connectors直接以Debezium作为底层把「订阅Binlog」这件事封装成了一个Flink Source上层拿到的是已经结构化的变更事件。我一般会这么理解基于Binlog的CDC不是在轮询数据库而是在消费一款数据库本身就在生产、本来就是为了复制而生的消息队列。这不是创新而是顺应了数据库已有的设计。2.3 一张对比表看清Flink为什么要选Binlog对比维度基于查询的CDC基于Binlog的CDC开源产品Sqoop、DataXCanal、Maxwell、Debezium执行模式BatchStreaming能否捕获所有数据否是变化延迟高延迟低延迟是否增加数据库压力是否为什么「能否捕获所有数据」这一栏差这么大根源就在DELETE。基于查询的方案只能发现「还在表里且字段有变化」的行删掉的行会无声消失UPDATE到旧值也无法区分。基于Binlog的方案则不同Binlog记录的是事件本身INSERT是一个事件UPDATE包含before和after两个镜像DELETE也作为一个完整事件落盘。Flink CDC消费这些事件时更新和删除都是显式的目标端可以对每一条做精确的幂等处理。日志方案还带来一个隐含红利位点offset。既然数据是一条事件流消费者就可以记录自己读到了哪一条下次从那个位置接着读。这就是Flink CDC能够做断点续传的根基。基于查询的方案没有位点概念每次轮询都是「从零开始找变化」自然也没有「从上次失败的地方继续」这回事。搞懂这一点后面讲Checkpoint和Savepoint时才不会觉得那是额外附加的复杂机制。3. Flink CDC 3.0的启动策略initial、earliest、latest背后如何拼接全量与增量把连接器建出来只是第一步。真正决定任务行为的是启动策略任务第一次启动时到底从哪一刻开始读数据这个问题比看起来复杂得多因为Binlog只包含「开启日志之后」的变更表里已经存在的历史数据它管不着。想要一份完整的数据就必须把「当前全量快照」和「快照之后的Binlog」拼成一条无缝的流。Flink CDC的StartupOptions就是这五种拼接方式的入口。3.1 五种启动模式五条进入Binlog的路径讲义里DataStream示例代码把最核心的构建过程浓缩成了这一段MySqlSourceString mySqlSource MySqlSource.Stringbuilder() .hostname(hadoop103) // MySQL地址 .port(3306) // MySQL端口 .databaseList(test) // 捕获的库 .tableList(test.t1) // 捕获的表 .username(root) .password(000000) .deserializer(new JsonDebeziumDeserializationSchema()) // 转成JSON字符串 .startupOptions(StartupOptions.initial()) // 先全量快照再无缝接增量 .build();builder里每一行参数都对应一个能力开关databaseList和tableList决定捕获范围deserializer决定下游拿到什么格式而startupOptions决定任务的起点。注意databaseList写的是库名tableList写的是「库.表」的完整形式写错一个字符任务起来后可能什么都读不到。Five种启动模式分别对应五条不同的进入路径initial首次启动先对监听表做一次全量快照快照完成后再从那个时间点继续读Binlog。最常用的模式也是「全量增量」语义的正确实现。earliest不做快照直接从Binlog里能找到的最早位置开始读。听起来很诱人但前提是Binlog要保留完整的数据库历史而MySQL的expire_logs_days默认只保留几天几乎没有生产库能满足这个条件。latest不做快照只从任务启动时刻的最新位点开始读启动之前的变更一概不管。适合「只要新数据」的场景但它补不了历史数据。specificOffset不做快照直接指定Binlog文件名和偏移量开始读。适合从已知位点精确恢复前提是你确实知道那个位点意味着什么。timestamp不做快照从指定时间戳之后的变更开始读。需要从Binlog开头遍历到第一个满足条件的事件成本比specificOffset更高。五种模式的选择本质上是回答一个问题目标端需要的是「从零开始的完整数据」还是「只关心启动之后的增量」。选错模式的后果不是报错而是「少数据」这种更难排查的故障后面避坑章节会专门讲。3.2 initial到底做了什么快照与增量的拼接点initial之所以成为默认推荐是因为它最接近人对「同步」的直觉先把现状搬过去再把之后的变动持续送过去。但「搬现状」和「读日志」之间有一个时间缝隙如何不让这个缝隙里的数据丢失是initial实现里最精巧的部分。常见的实现思路是启动时先记录一个Binlog位点然后并发地把表数据按主键分片做全量快照。快照期间持续发生的变更不会被丢弃它们会进入增量缓冲区等全量快照全部完成后连接器从之前记录的那个位点开始消费Binlog追上快照期间的变更再切到纯增量模式。先取位点再快照是为了给快照一个明确的时间边界如果反过来先快照再取位点快照数据和位点之间就会出现一段灰色的空洞。位点先行的方式会带来少量重复读但重复比丢失安全得多。这也是为什么快照阶段要开Checkpoint。全量快照的进度、每个分片读到哪了、Binlog消费到什么位置这些信息全部要以状态的形式存进Checkpoint。任务崩溃或重启时Flink从Checkpoint恢复才能知道「快照读到一半的分片继续读而不是从头再扫一遍」。我一般会这么理解initial不是两个阶段的简单串联而是「快照进度」和「日志位点」两条状态线在Checkpoint里的对齐。讲义里强调开启Checkpoint、设置ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION都是在为这个对齐兜底。3.3 一条变更如何变成JSON反序列化机制的底层DataStream示例里用了JsonDebeziumDeserializationSchema把Debezium的SourceRecord转成JSON字符串。这一步转换在print()里看起来只是「多了一行输出」实际上下游所有逻辑都依赖这个JSON的字段结构。一条UPDATE事件从MySQL到Flink被翻译成的JSON大致长这样{ before: { id: 1001, name: zhangsan }, after: { id: 1001, name: zhangsan_new }, op: u, ts_ms: 1735000000000 }op字段标记事件类型r代表快照阶段读出来的历史数据c是INSERTu是UPDATEd是DELETE。快照阶段读到的一行会以opr出现很多新手在这里被迷惑以为这是业务数据里真实存在的一种「操作类型」。实际上它只是连接器在告诉你「这条来自快照不是Binlog增量」。INSERT事件里before为nullDELETE事件里after为nullUPDATE事件则同时携带旧值和新值。理解这个结构对排查问题特别有用怀疑数据被错误覆盖时去看before和after的差异怀疑删除没同步时去看有没有opd的事件怀疑全量增量衔接有问题时去看是不是快照阶段后op字段一直停在r没有切到c/u/d。JSON格式没有把事件类型隐藏在某个二进制结构里所有信息都对下游可见这也是DataStream方式调试起来最顺手的原因。4. Flink CDC的常见问题与避坑五个真实翻车现场从开Binlog到恢复位点讲义把DataStream、FlinkSQL、pipeline三种方式完整跑了一遍对照复现时最容易翻车的点其实集中在少数几个环节。它们看起来各自独立根源却都在「日志没开、位点没存、身份冲突、时区错位」这四类问题上。以下是五个典型的踩坑现场按现象到原因再到解决的顺序拆开说。4.1 数据一条都读不到不是代码问题是Binlog没开现象任务正常启动print()没有输出任何数据MySQL里明明有表有数据连接器就像什么都没发生一样。原因MySQL的my.cnf里根本没配log-bin或者binlog_format不是row。Flink CDC读取的是Binlog事件不是直接查表日志没开连接器就没有数据来源。binlog-do-db如果没包含目标库同样会导致只连得上、读不到。这类问题不会报错只表现为「静默无数据」最容易浪费时间。解决按讲义配置server-id1、log-binmysql-bin、binlog_formatrow再用binlog-do-db指定要同步的库改完必须重启MySQL。启动后用SHOW MASTER STATUS确认Binlog已经开启、格式是ROW再重新提交Flink任务。这一步是后续所有操作的前提值得在每次环境初始化时核对一遍。4.2 数据串表、连接报错server-id冲突比想象中常见现象两个Flink任务同时连同一个MySQL实例运行一段时间后日志开始报连接异常目标端数据出现串表或重复。原因每个基于Binlog的CDC连接器在MySQL眼里都是一台「从库」而server-id就是它的从库身份标识。两个任务如果配了同一个server-idMySQL会把它们当成同一个客户端Binlog分发就会错位。这个问题在单任务时完全看不出来一旦并行跑多个任务就爆发看起来很像玄学其实是身份标识冲突。解决每个任务使用独立的server-id并发度高时给一个区间比如5400-5404。区间长度要覆盖任务内可能建立的复制连接数实践上多留几个id更稳。这也是第5章YAML里server-id要写成一个范围的原因——它不是可选项是身份隔离的必需品。4.3 全量都进来了增量就是不动启动模式选错了现象一启动任务历史数据全部同步到位看起来一切正常但随后往MySQL里插入新数据目标端毫无反应。原因startupOptions配的是latest而不是initial。latest模式明确不读启动前的Binlog启动之后才消费日志如果你的意图是「先全量再增量」就一定要用initial。另一种隐蔽的情况是Checkpoint没开任务重启后位点清零看起来就像增量断了。它和上面的模式选错有着相同的表象但解决路径完全不同。解决首次接入需要完整数据时用StartupOptions.initial()它内部完成了快照和增量的衔接。同时确认Checkpoint已开启state后端和存储路径可写否则位点无法持久化重启即丢。记住一句话增量不动的排查顺序先看启动模式再看Checkpoint。4.4 从Savepoint恢复却重读全量状态后端和存储路径不一致现象按讲义的流程创建Savepointcancel掉任务修改MySQL数据后从Savepoint重启日志显示又开始做全量快照而不是从上一次的Binlog位点接着读。原因Savepoint能恢复的前提是任务的状态里保存了Binlog位点和快照进度。如果代码里设置的状态后端是HashMapStateBackend而集群默认状态后端是RocksDB或Filesystem恢复时状态对不上或者Savepoint的存储路径和CheckpointStorage指向的不是同一个位置Flink找不到有效状态就会退化成从空状态启动于是重新全量快照。解决保证三处一致——代码里的setStateBackend、CheckpointConfig里的setCheckpointStorage、执行恢复时的savepoint路径。恢复命令用bin/flink run -s hdfs://.../savepoint-xxx指定同一个HDFS路径启动后先看日志里是否出现了「恢复状态」的相关输出再判断是否接上了位点。4.5 同步到Doris的数据差8小时、DDL跟不上时区和建表属性没配现象数据能同步到Doris但时间字段比MySQL里的值晚了或早了8小时MySQL执行ALTER TABLE ADD COLUMN后Doris目标表没有出现新列。原因server-time-zone没有显式配置连接器按MySQL服务器默认时区解析DATETIME跨时区部署就会出现8小时偏差。Doris侧的问题通常是sink配置里没有开启light_schema_changeDDL变更走了全量重写或直接被拒绝。这两个问题都不会让任务失败只会让数据「看起来不对」属于最难察觉的一类故障。解决source配置里显式写server-time-zone: UTC8保证连接器、Flink、Doris三端时区一致。sink配置里加table.create.properties.light_schema_change: true让Doris用轻量Schema变更响应MySQL的DDL测试环境再配replication_num: 1避免多副本资源不足。改完配置后用一条ALTER TABLE ADD COLUMN作为验收动作验证DDL确实能同步过去。提示遇到「同步偶尔断、重启就重来」这类问题不要先怀疑代码逻辑按「Binlog格式→server-id区间→Checkpoint是否开启→时区和DDL属性」这条链路逐层排查多数坑都藏在这四个环节里。5. 从MySQL到Doris的Streaming ETL一个YAML文件如何编排整条同步链路Flink CDC 3.0引入的pipeline模式把前面说的所有机制打包成了一个更上层的产品形态不需要写Java代码不需要创建Source和Sink一份YAML描述清楚「从哪读、写到哪、怎么跑」剩下的交给框架。DataStream和FlinkSQL仍然是可用的但pipeline模式明显在向「同步工具」的方向收敛这也是MySQL到Doris这类Streaming ETL场景最顺手的接入方式。5.1 一个YAML文件三块配置source、sink、pipeline讲义里mysql-to-doris.yaml的内容结构非常清晰source: type: mysql hostname: node01 port: 3306 username: root password: 123 tables: test.\.* # 匹配test库下的所有表 server-id: 5400-5404 # 模拟从库的server-id区间 server-time-zone: UTC8 # 显式指定会话时区 sink: type: doris fenodes: node01:7030 username: root password: 000000 table.create.properties.light_schema_change: true # 开启轻量Schema变更 table.create.properties.replication_num: 1 # 单副本测试环境 pipeline: name: Sync MySQL Database to Doris parallelism: 13三块配置分别对应三个层次source描述数据从哪来sink描述数据写到哪pipeline描述整个任务以什么名字和并行度运行。source里的tables用了一个正则表达式test.\.*它匹配test库下的所有表这样新增表时不需要改配置。server-id区间5400-5404给任务预留了5个身份标识供快照分片和增量消费阶段领取。sink部分的fenodes指定Doris FE的地址table.create.properties是自动建表时透传给Doris的属性light_schema_change开启轻量Schema变更让加列、改类型等DDL走快速通道replication_num指定副本数单机测试环境必须设为1否则多副本会因为节点不足而建表失败。pipeline.parallelism决定整条管线的算子并行度。这里要注意source端实际建立的MySQL复制连接数还要受server-id区间长度约束区间太小而并行度过高任务在快照阶段就可能报身份冲突。把这份YAML和DataStream方式对比着看会发现它把反序列化格式、位点管理、启动策略全部封装成了默认行为。不需要写JsonDebeziumDeserializationSchema不需要配置CheckpointStorage不需要处理SourceRecord到业务对象的映射——这些被抽象进了pipeline内部对使用者来说是一个黑匣子。黑匣子带来易用性也意味着你必须信任它的默认选择所以配置里的每一项默认值反而值得多看两眼。5.2 正则匹配、自动建表与DDL同步三个细节的语义tables字段的正则是这份配置里最容易被低估的一行。test.\.*在YAML里写成这样是因为反斜杠本身需要转义实际传给正则引擎的是test\..*含义是「库名test后面跟一个点再跟任意字符」。用它替代具体表名列表换来的是新表自动纳入同步范围。如果你的同步范围要跨多个库可以写多个正则或调整库名部分如果只想同步某几张表就明确写表名避免误伤。自动建表则把「目标表不存在」这个常见故障从运维清单里划掉了。任务启动时source侧读取到的表结构会经过schema推导转换成Doris的建表语句执行。这背后依赖Flink CDC对MySQL表结构的实时感知不仅启动时读一次运行期间MySQL执行ALTER TABLE连接器也会捕获DDL事件并转发给Doris端。这就是为什么light_schema_change必须开启——它让Doris能以低代价响应频繁的加列操作否则每次DDL都可能触发表的长时间重写。DDL同步这件事是pipeline模式相对DataStream手写方案最明显的优势。自己用DataStream做同步时Binlog里的DDL事件需要手工解析Doris端的表结构变更需要自己拼SQL执行等于把一套Schema管理逻辑重写一遍。pipeline模式把这些收编成配置项代价是你失去了对每个环节的细粒度控制但这种交换在「MySQL到Doris」这种同构同步场景下是划算的。5.3 启动演练Checkpoint开好、Doris建好库、提交任务讲义里这套流程的启动顺序值得原样保留。第一步是改Flink配置在flink-conf.yaml里加一行execution.checkpointing.interval: 5000让Flink每5秒做一次Checkpoint。pipeline模式下的位点恢复依赖这套机制不开Checkpoint任务异常后只能重新全量同步前面的配置再完整也白搭。改完配置分发到所有节点再bin/start-cluster.sh拉起集群。随后启动Doris的FE和BE并在Doris里先建好目标库create database test; create database doris_test_route;这一步容易被跳过。pipeline会自动建表但不会自动建库库必须事先存在。建好后提交同步任务bin/flink-cdc.sh job/mysql-to-doris.yaml任务启动后用MySQL客户端往test库的任意表插入、修改数据再刷新Doris对应表能看到行数据实时变化。更关键的验收动作是执行一次ALTER TABLE ADD COLUMN刷新Doris端表结构确认新列出现了。如果这一步通过说明整条链路从Binlog捕获、schema推导、DDL同步到目标端写入全部打通。要验证位点恢复可以在任务运行片刻后直接杀掉进程再从最近一次Checkpoint位置重启Flink CDC任务。正常情况下它会接着上次的Binlog位点继续消费而不是重新做全量快照。这和DataStream方式依赖Savepoint恢复是同一套状态机制只是触发方式更隐蔽——pipeline把「状态在哪、什么时候恢复」的细节全部藏了起来你看不到但它确实在工作。6. 设计取舍声明式pipeline换掉手写Source值不值DataStream、FlinkSQL、pipeline三种方式复杂度从高到低控制力也从强到弱。DataStream能精确控制反序列化逻辑、自定义过滤和转换算子但代价是每张表都要写一遍builder参数DDL同步还要自己解析FlinkSQL用建表语句声明Source和Sink简洁但多表、动态表场景不够灵活pipeline模式把整个链路浓缩成一份YAML换来的是自动建表、正则匹配、DDL同步和位点管理开箱即用。这是一条很清晰的取舍曲线把「从MySQL读到Doris写入」固定成一条强约束的同步管线框架替你承担复杂度同时替你做出一系列默认决策。哪条路更好取决于你的场景。如果你做的是一次性的历史数据回填DataStream足够如果目标是持续的多表整库同步pipeline的声明式配置远比手写代码容易审计和维护。配置文件本身就是文档任何同事看到这份YAML都能说清楚数据从哪来往哪去而手写代码还得先跑通一次才能确定行为。但pipeline也有限制复杂的清洗逻辑、多流join、自定义的状态管理它都表达不了这些场景最终还是要回到DataStream。我自己的习惯是先明确这个同步任务是要「搬数据」还是「算数据」。只搬不算是pipeline的主场一旦涉及计算就老老实实写DataStream。这份讲义把三种接入方式完整走了一遍运行日志、SQL输出、配置文件都保留着适合在自己集群里逐步复现。从那以后我每次做MySQL到Doris的同步都会强制走一遍先确认binlog_formatrow再分配独立server-id区间然后开Checkpoint最后用一次ALTER TABLE ADD COLUMN作为验收动作。这套流程多半是用血泪经验换来的希望帮到你。本文还有配套的精品资源点击获取