Flink CDC 3.0实战：MySQL到Doris实时同步全攻略-北京尧图网络

简介面向大数据实时同步场景的 Flink CDC 3.0 实战资料由尚硅谷研究院编写适合有一定 Flink 基础的研发工程师目标是解决 MySQL 到 Doris 的高效稳定数据同步问题。资料先厘清 CDC 定义对比基于查询与基于 Binlog 两种方式的差异突出 Binlog 方案可捕获全部变更、延迟低且不增加数据库压力并介绍 flink-cdc-connectors 组件能力。核心部分给出完整案例实操包含 MySQL 数据准备、DataStream 与 Flink SQL 两种接入方式以及 MySQL 到 Doris 的 Streaming ETL 实现覆盖环境搭建、配置编写、任务启动与结果测试附详细代码示例和操作指南。资源为 1 个 docx 文档大小 145KB内容紧凑、步骤清晰重点提示开启 Binlog、设置检查点等可靠性配置便于读者按需调参并直接应用于项目。目前已有 1788 人学习下载适合需要快速上手 Flink CDC 同步链路的工程师。1. Flink CDC 3.0这份资料把 MySQL 到 Doris 的实时同步讲透了做大数据开发的人应该都有过这种体验业务库的 MySQL 数据要同步到数仓以前要么用 Sqoop 跑离线批量要么用 DataX 定时拉延迟高不说还总给业务库增加压力。等你想上实时同步的时候又得在 Canal、Maxwell 之间纠结搭一套 Debezium Kafka 的链路光维护就够喝一壶的。这份 Flink CDC 3.0 的资料目标就是解决这个痛点——它把 MySQL 的 Binlog 捕获、解析、分发到最终写入 Doris 的整条链路完整演示了一遍既有 DataStream API 的底层写法也有 FlinkSQL 的声明式用法还包含了 3.0 版本主推的 YAML 配置化同步方案。我拆完这份文档的结论是对于想从零搭建一套 MySQL 实时数仓同步管道、或者正在调研 CDC 技术的工程师来说这里面所有的代码和配置都是可以照着敲的踩坑点也写得比较明确。2. CDC 技术选型与项目架构为什么最终落在 Flink 上2.1 两种 CDC 实现方式的本质差异文档第一章花了不小篇幅对比基于查询和基于 Binlog 两种 CDC 方案的差异。这个对比值得细看因为它决定了同步管道的整体架构选型。对比维度基于查询的 CDC基于 Binlog 的 CDC代表产品Sqoop、DataXCanal、Maxwell、Debezium执行模式Batch 批量Streaming 流式能否捕获所有数据变更否只能捕获查询时刻的快照是能捕获插入、更新、删除全部操作延迟性高延迟取决于调度周期低延迟毫秒级对数据库的压力增加压力频繁查询几乎无压力基于 Binlog 复制协议基于查询的方式本质上是定期对源库做全量或增量查询然后比对数据差异。这种方式有两个先天缺陷一是无法感知删除操作二是性能瓶颈明显。基于 Binlog 的方式则是让 MySQL 自己把变更记录写到二进制日志里CDC 工具伪装成从库去拉取 Binlog然后按事件类型解析。这种方式拿到的是一条完整的、带事务语义的变更流可以精确还原每一次插入、更新和删除。这个差异直接影响了架构决策如果你只需要每天凌晨同步一次全量数据用 DataX 就够了但如果业务需要分钟级甚至秒级的实时数仓Binlog 方案就是唯一选择。Flink 社区正是因为看中了这个趋势才在 flink-cdc-connectors 组件里直接内置了 MySQL、PostgreSQL 等数据库的 CDC source让这批数据能以流的方式进入 Flink 计算引擎再通过丰富的 sink 组件落到 Doris、StarRocks、Kafka 等目标端。2.2 Flink 在 CDC 生态里的位置与版本演进Flink CDC 组件之所以能流行是因为它把「读取 Binlog 变更流」这件事封装成了一个标准的 Flink source connector。你在 DataStream API 里把它当成一个数据源接入在 FlinkSQL 里注册一张映射表它就能持续不断地往 Flink 作业里推送数据变更事件。其中每条事件记录都包含操作类型op 字段、变更前后的数据内容以及 Binlog 的位点信息。版本演进上Flink CDC 的核心能力正在从「主库连接器」向「同步框架」转变。2.x 版本你需要自己维护整条链路先建表、写代码、再配置 sink每一层都需要自己完善。3.0 版本最大的变化是提供了一套 YAML 配置化同步方案你只需要写一个描述 source、sink、pipeline 的文件就能启动一条完整的同步管道。更关键的是引入了路由机制可以灵活地把源端表映射到目标端表这在多库多表同步场景下意义重大。架构上可以分为三层去理解最底层是 Debezium 嵌入式引擎负责真正的 Binlog 解析和变更事件捕获中间层是 Flink CDC connector 对 Debezium 事件的封装和适配最上层就是用户直接使用的 API 层包括 DataStream 模式、SQL 模式以及 YAML pipeline 模式。资料提供的案例恰好把这三层都覆盖到了从最底层的 API 用法到最高层的配置化同步一应俱全。2.3 项目目录结构与前置环境需求按资料中的实践场景梳理这套案例涉及的核心环境组件如下MySQL8.0.x源数据库测试表包含 t1、t2、t3 三张表分布在 test 和 test_route 两个库Flink1.18.0流式计算引擎同时承担状态存储与 Checkpoint 管理Flink CDC3.0.0提供 MySQL source connector 和 Doris pipeline connectorDoris1.2.x目标数据库作为实时数仓的分析引擎HDFS用于存储 Flink Checkpoint 和 Savepoint 的状态数据整个项目的文件组织也围绕这些组件展开Maven 工程里包含 flink-java、flink-streaming-java、flink-clients、flink-table-planner、flink-table-runtime 等 Flink 基础依赖以及 flink-connector-mysql-cdc 核心连接器和 mysql-connector-java 驱动独立部署的 flink-cdc-3.0.0 目录下需要手动放置 flink-cdc-pipeline-connector-doris-3.0.0.jar 和 flink-cdc-pipeline-connector-mysql-3.0.0.jar用于支撑 YAML 配置化同步任务。3. DataStream 模式实操从环境准备到断点续传完整跑通3.1 开启 MySQL Binlog 与初始数据准备在写任何代码之前先把 MySQL 的 Binlog 打开。文档给了一个很清晰的配置方案在 /etc/my.cnf 里追加如下配置# 数据库id server-id 1 # 启动binlogmysql-bin是binlog的文件名前缀 log-binmysql-bin # binlog类型maxwell要求为row类型 binlog_formatrow # 启用binlog的数据库需根据实际情况作出修改 binlog-do-dbtest binlog-do-dbtest_route注意几个关键点server-id 必须设置这是 MySQL 主从复制协议中每个实例的唯一标识binlog_format 必须为 row因为只有 row 格式的 Binlog 才记录了每一行数据变更前后的完整镜像而 statement 格式只记录 SQL 语句无法还原精确的变更数据binlog-do-db 建议按需指定只对需要的库开启 Binlog 可以减少日志量避免对数据库性能造成不必要的额外影响。binlog-do-db 这个参数容易踩坑。它只能精确到数据库级别如果你后续新增了一个也需要同步的库必须记得改配置并重启 MySQL。另外还有个隐含条件——数据表本身必须存在主键如果表没有主键CDC 解析出的落库操作无法准确匹配变更行容易造成数据错乱。数据准备阶段需要在 test 数据库里建 t1、t2、t3 三张表并插入测试数据然后在 test_route 数据库里建同样结构的表但插入不同的数据。这个设计是为了后面验证「同步变更」和「路由变更」两种场景区分哪些表是直接映射同步、哪些表需要做库名或表名的路由转换。3.2 Maven 工程搭建与 Flink CDC 依赖引入使用 Maven 管理工程参考资料中的依赖结构如下properties maven.compiler.source8/maven.compiler.source maven.compiler.target8/maven.compiler.target flink-version1.18.0/flink-version project.build.sourceEncodingUTF-8/project.build.sourceEncoding /properties dependencies dependency groupIdorg.apache.flink/groupId artifactIdflink-java/artifactId version${flink-version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink-version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version${flink-version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-table-planner_2.12/artifactId version${flink-version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-table-runtime/artifactId version${flink-version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-table-api-java-bridge/artifactId version${flink-version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-base/artifactId version${flink-version}/version /dependency dependency groupIdcom.ververica/groupId artifactIdflink-connector-mysql-cdc/artifactId version3.0.0/version /dependency dependency groupIdmysql/groupId artifactIdmysql-connector-java/artifactId version8.0.31/version /dependency /dependencies build plugins plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-assembly-plugin/artifactId version3.0.0/version configuration descriptorRefs descriptorRefjar-with-dependencies/descriptorRef /descriptorRefs /configuration executions execution idmake-assembly/id phasepackage/phase goals goalsingle/goal /goals /execution /executions /plugin /plugins /build这里的依赖选择有讲究。flink-java 和 flink-streaming-java 是流处理的基础flink-table-planner 和 flink-table-api-java-bridge 是后续 FlinkSQL 模式必需的即使当前只用 DataStream API也建议提前引入。打包插件用 maven-assembly-plugin 的 jar-with-dependencies 模式能把所有依赖打进一个可执行的 fat jar这样提交到 Flink 集群时不需要额外分发依赖包。flink-connector-mysql-cdc 3.0.0 的 groupId 是 com.ververica注意不要写成 org.apache.flink。mysql-connector-java 用于底层连接 MySQL 实例。3.3 核心代码逐段拆解MySqlSource 构建与执行DataStream 模式的核心代码可以按逻辑拆成几个独立模块。先看环境准备和 Checkpoint 配置// TODO 1. 准备流处理环境 StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // TODO 2. 开启检查点Flink-CDC将读取binlog的位置信息以状态的方式保存在CK // 如果要做到断点续传需要从Checkpoint或者Savepoint启动程序 // 2.1 开启Checkpoint每隔3秒钟做一次CK并指定CK的一致性语义 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); // 2.2 设置超时时间为1分钟 env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); // 2.3 设置两次重启的最小时间间隔 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); // 2.4 设置任务关闭的时候保留最后一次CK数据 env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 2.5 指定从CK自动重启策略 env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L))); // 2.6 设置状态后端 env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage(hdfs://hadoop102:8020/flinkCDC); // 2.7 设置访问HDFS的用户名 System.setProperty(HADOOP_USER_NAME, atguigu);Checkpoint 对 Flink CDC 来说不是可选配置而是必选配置。原因在于 Flink CDC source 读取 Binlog 时会把当前消费到的 Binlog 位点信息存放在 Flink 的状态State中。如果没有开启 Checkpoint状态不持久化作业一重启就要重新读全量数据开启后作业崩溃时能从最近一次 Checkpoint 恢复位点真正做到断点续传。建议保持每 3 到 5 秒一次 Checkpoint 的频率太频繁会增加 HDFS 的写入压力太稀疏则可能丢失较多数据。Checkpoint 存储路径配置到了 HDFS这意味着在本地 IDE 里直接跑这段代码会报错必须先启动 HDFS 集群。文档给出的hdfs://hadoop102:8020是一个标准配置你在自己环境里需要替换为实际 NameNode 的地址和端口。RETAIN_ON_CANCELLATION 也很关键它保证你在 WebUI 上手动取消作业后最后一次 Checkpoint 仍然保留在 HDFS 上后续可以从这个保留的 Checkpoint 恢复位点。再来看 MySqlSource 的构建MySqlSourceString mySqlSource MySqlSource.Stringbuilder() .hostname(hadoop103) .port(3306) .databaseList(test) // 设置捕获的数据库 .tableList(test.t1) // 设置捕获的数据表 .username(root) .password(000000) .deserializer(new JsonDebeziumDeserializationSchema()) // 将SourceRecord转成JSON字符串 .startupOptions(StartupOptions.initial()) .build();builder 模式的可读性很好。hostname 和 port 指定了 MySQL 地址databaseList 用逗号分隔多个库名tableList 的格式是库名.表名同样支持逗号分隔多个表。deserializer 选择 JsonDebeziumDeserializationSchema则每条变更记录会以 JSON 格式输出包含 binlog 文件名、位点偏移、操作类型、数据内容等字段。startupOptions 决定了作业启动时从什么位置开始读取initial如果表里已有数据先做一次全量快照再继续读最新的 Binlog 变更。这是最常用的选项兼顾历史数据和实时增量earliest不做全量快照直接从 Binlog 最早的位置开始读。使用前提是 Binlog 保留了完整的变更历史latest不做全量快照只从当前时刻开始读增量作业启动前的数据变更不会被捕获specificOffset从一个指定的 Binlog 位点开始读适用于从历史备份恢复场景timestamp从某个时间点之后的 Binlog 开始读最后一个关键部分是接入和执行// TODO 4.使用CDC Source从MySQL读取数据 DataStreamSourceString mysqlDS env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), MysqlSource); // TODO 5.打印输出 mysqlDS.print(); // TODO 6.执行任务 env.execute();将构建好的 mySqlSource 通过 env.fromSource 接入流处理环境WatermarkStrategy.noWatermarks() 表示这个数据源不需要分配水位线因为 CDC 数据本身没有事件时间的处理需求用处理时间即可。print() 会把每条变更事件以 JSON 字符串形式输出到 TaskManager 日志用于验证捕获效果。这样一个最简单的 CDC 作业就完成了打包提交到 Flink 集群就能看到 MySQL 里 test.t1 表的每次增删改都会以 JSON 格式输出一条记录。3.4 Savepoint 恢复与断点续传验证流程文档在测试部分给出了一个完整的「创建 Savepoint → 重启作业 → 验证断点续传」流程。这里简要梳理一下其中的操作逻辑# 给作业创建SavepointJobId在WebUI上可以找到 bin/flink savepoint JobId hdfs://hadoop102:8020/flinkCDC/save # 在WebUI上cancel掉当前作业 # 在MySQL的test.t1表中执行insert、update、delete操作 # 从Savepoint重启作业 bin/flink run -s hdfs://hadoop102:8020/flinkCDC/save/savepoint-5dadae-02c69ee54885 \ -c com.atguigu.cdc.FlinkCDCDataStreamTest ./gmall-flink-cdc.jar这段操作验证的是整个 CDC 管道的可靠性作业从 Savepoint 启动后会跳过已经消费过的 Binlog 事件直接从保存的位点继续读取。你在作业停止期间对 MySQL 做的 DML 操作重启后都会被正确捕获不会重放历史数据也不会丢数据。4. FlinkSQL 模式应用CDC 源表化与实时链路简化4.1 用 CREATE TABLE DDL 声明 CDC 数据源DataStream 模式虽然灵活但需要写不少代码而且对上游表结构的变化不敏感。FlinkSQL 模式则把 CDC source 声明成一张维表用 DDL 的方式描述表结构、连接参数和读取模式开发效率会高不少。资料中给出的示例代码如下package com.atguigu; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.Table; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; public class FlinkSQLTest { public static void main(String[] args) { // 创建流执行环境和表执行环境 StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); StreamTableEnvironment tableEnv StreamTableEnvironment.create(env); // 通过DDL声明一张MySQL CDC映射表 tableEnv.executeSql( create table t1( id string primary key NOT ENFORCED, name string ) WITH ( connector mysql-cdc, hostname hadoop103, port 3306, username root, password 000000, database-name test, table-name t1 ) ); // 查询表中数据 Table table tableEnv.sqlQuery(select * from t1); table.execute().print(); } }这段代码背后的原理值得说明当你在 FlinkSQL 里注册这张表时系统会自动根据 connector mysql-cdc 定位到对应的 source connector并根据 database-name 和 table-name 去 MySQL 中读取对应的表结构和数据。在声明字段时id 被标记为 primary key NOT ENFORCED意思是声明了主键但不由 Flink 本身强制校验主键的作用主要是供精确一次处理时识别重复记录。FlinkSQL 模式相比 DataStream 模式的好处是省掉了反序列化工作JsonDebeziumDeserializationSchema 的行为被内置封装了。你查询出来的数据直接就是对应的 Java/SQL 类型不再需要手动解析 JSON。对于复杂的数据加工场景可以直接在 SQL 里做 join、聚合、窗口计算语法生态远比 DataStream API 便宜。4.2 从 SQL 到实时数仓 ODS 层的封装思路实际生产环境中 FlinkSQL 模式最常见的用途是把 CDC 数据接入实时数仓 ODS 层。一个典型的做法是在 FLinkSQL 里声明 MySQL CDC 源表然后创建一张 Kafka sink 表通过一条 INSERT INTO 语句把变更数据持续写入 Kafka topic下游再通过 Kafka 数据源接入其他组件做加工处理。-- 声明上游MySQL CDC表 CREATE TABLE ods_mysql_t1 ( id STRING PRIMARY KEY NOT ENFORCED, name STRING ) WITH ( connector mysql-cdc, hostname hadoop103, port 3306, username root, password 000000, database-name test, table-name t1, scan.startup.mode initial ); -- 声明下游Kafka sink表 CREATE TABLE kafka_t1 ( id STRING, name STRING, op_type STRING ) WITH ( connector kafka, topic ods_t1, properties.bootstrap.servers hadoop102:9092, format json ); -- 将变更数据写入Kafka INSERT INTO kafka_t1 SELECT id, name, UPSERT FROM ods_mysql_t1;这种设计思路是把 CDC 的实时性优势真正利用起来。与离线数仓每天一次全量同步不同CDC 让 ODS 层能从分钟级延迟降到秒级延迟而且天然保留了数据的变更历史。你可以通过查询 Binlog 中的 op 字段来判断一条记录是 INSERT、UPDATE 还是 DELETE这为后续的增量数仓建模提供了重要依据。5. MySQL 到 Doris 的 Streaming ETL 核心应用YAML 配置化同步实战5.1 环境准备与应用部署前面两个章节讲的是编程 API 的用法而 Flink CDC 3.0 最吸引人的新特性其实在 YAML 配置化同步这里。它把常见的数据同步场景从「写代码」直接变成「写配置文件」这在小规模团队或者快速交付场景下非常实用。准备阶段分三步走# 1. 解压Flink CDC 3.0安装包到目标目录 tar -zxvf flink-cdc-3.0.0-bin.tar.gz -C /opt/module/ # 2. 将MySQL和Doris的pipeline连接器放入lib目录 cp flink-cdc-pipeline-connector-doris-3.0.0.jar /opt/module/flink-cdc-3.0.0/lib/ cp flink-cdc-pipeline-connector-mysql-3.0.0.jar /opt/module/flink-cdc-3.0.0/lib/ # 3. 启动Doris FE和BE bin/start_fe.sh bin/start_be.sh # 4. 在Doris中创建目标数据库 mysql -uroot -p000000 -P9030 -hhadoop102 CREATE DATABASE test; CREATE DATABASE doris_test_route;有几个细节需要强调。flink-cdc-pipeline-connector-doris 和 flink-cdc-pipeline-connector-mysql 这两个 jar 必须手动放进 flink-cdc-3.0.0 安装包的 lib 目录因为官方发布包默认不包含特定数据库的 pipeline 连接器。Doris 的前端端口是 9030FE 端口是 8030HTTP 端口是 8040很多人会把 9030 和 8030 混在一起导致客户端连接失败。Doris 本身兼容 MySQL 协议所以可以使用 MySQL 命令行客户端直接访问。5.2 YAML 配置文件逐行解析source、sink、pipelineFlink CDC 3.0 的 YAML 同步配置从结构上分成 source、sink、pipeline 三大块。先看同步变更场景的配置source: type: mysql hostname: node01 port: 3306 username: root password: 123 tables: test.\.* server-id: 5400-5404 server-time-zone: UTC8 sink: type: doris fenodes: node01:7030 username: root password: 000000 table.create.properties.light_schema_change: true table.create.properties.replication_num: 1 pipeline: name: Sync MySQL Database to Doris parallelism: 1source 部分的核心参数是 tables, 这里的test.\.*是一个正则表达式表示匹配 test 库下的所有表。如果你只想同步特定表可以写成test.t1或test.t[0-9]这种正则匹配的灵活性在多表同步场景里非常实用。server-id 需要配置一个范围4000 到 5404 意味着可以同时开 5 个 Binlog 消费线程对于多表同步场景建议至少配置 2 个以上的 server-id 段避免表数量多时单一线程成为瓶颈。sink 部分的 fenodes 是 Doris 的 FE 节点地址。table.create.properties.light_schema_change开启后Doris 建表时启用轻量级 Schema Change 特性当源端新增列时目标端无需重写数据文件只需做元数据变更这个参数在业务表结构经常调整的场景下建议打开。replication_num: 1则在 BE 上只保留一个副本如果你的 Doris 集群有多个 BE 节点且启用了高可用这里需要相应调大。pipeline 部分只需配置 name 和 parallelism前者用于展示同步任务名称后者控制并发度。这套配置对开发者的意义在于你在 MySQL 里对 test 库的任何 DML 或 DDL 变更都会经过 Flink CDC 管道的解析和转换自动落到 Doris 对应的表中。5.3 路由变更配置与多表映射实战路由变更场景是 Flink CDC 3.0 区别于 2.x 的重要能力。来看资料中的路由配置source: type: mysql hostname: hadoop103 port: 3306 username: root password: 000000 tables: test_route.\.* server-id: 5400-5404 server-time-zone: UTC8 route: - source-table: test_route.\.* sink-table: doris_test_route.\.*路由规则的本质是实现源表到目标表的映射转换。test_route.\.*和doris_test_route.\.*的映射关系是把 test_route 库下的所有表同步到 doris_test_route 库下。有了这个机制你可以在不修改业务库表结构的前提下对目标库的表名、库名做统一规划——比如把生产环境的 test_route 同步到数仓的 ods_test_route。在实际项目里基于这套理念你可以做不少有意思的事情。同一张 MySQL 表可以同时配置两条管道分别写入 Doris 的不同表中实现一份数据多种用途可以通过路由把多个 MySQL 实例的表汇聚到同一个 Doris 表前提是表结构一致也可以在目标端通过额外的 schema 变更逻辑实现增量列、删除列的自动映射。5.4 启动任务与 DDL 自动同步验证配置完成后启动命令很简洁# 启动Flink CDC同步变更任务 bin/flink-cdc.sh job/mysql-to-doris.yaml任务启动后Flink CDC 会先对指定的表做一次全量快照把已有数据全部写入 Doris然后监听 Binlog 持续同步增量变更。验证过程中可以在 MySQL 中执行新增列、修改列类型等 DDL 操作然后刷新 Doris 中的表结构观察 Schema 变化是否自动同步。这个能力对正在迭代中的业务尤为重要因为只要源端表加了新列同步管道能自动感知不需要手动在 Doris 里执行对应的 ALTER TABLE 语句。不过 DDL 同步也不是万能的。删除列的 DDL 在 Doris 端的支持可能不够完善如果线上执行 DROP COLUMN建议先在测试环境验证一遍再同步到生产管道。6. 常见问题排查与实验环境避坑记录6.1 MySQL 连不上Binlog 没开或 server-id 冲突这个是我在跑通案例时遇到的第一个问题。现象是我照着文档配好了 /etc/my.cnf 后重启 MySQL然后用 DataStream 方式启动 Flink 作业发现报错信息显示无法连接 MySQL或者连接后立刻被拒绝。原因有两类。一是很多 MySQL 安装包默认关闭了 Binlog如果你在 my.cnf 里只加了 log-bin 和 binlog_format 但没有加 server-idMySQL 可能仍然不写入 Binlog。二是 server-id 配置不能和另一个正在运行的 MySQL 实例相同如果集群里有多个 MySQL 实例server-id 重复会导致从库连接被踢掉。解决办法是在 my.cnf 的 [mysqld] 区段同时配置 server-id、log-bin、binlog_format 三个参数然后执行SHOW MASTER STATUS确认 Binlog 文件是否生成。确认没有问题后再启动 Flink CDC 作业这个问题就解决了。6.2 已提交 Flink 作业但看不到数据输出第二个问题也比较典型作业成功提交到 Flink 集群WebUI 上状态是 RUNNING,但 TaskManager 日志里完全看不到 print() 输出的数据。这种情况通常是表匹配正则没写好。CDC source 的 tableList 参数如果填了test.t1但实际 MySQL 中表名是T1或者表在另一个库中就一个事件都匹配不到。另外如果你使用了 databaseList 和 tableList 的组合Flink CDC 会按数据库名表名做联合匹配其中一个不匹配就拉不到数据。排查办法是先用初始状态做验证在 MySQL 里手动执行一条 INSERT 语句然后立刻观察 TaskManager 日志里是否有对应的 JSON 输出。如果没有去 WebUI 检查作业是否有反序列化异常同时确认日志中是否出现了连接 MySQL 的相关配置信息。我一般会先把 databaseList 和 tableList 的配置简化成只匹配一张表跑通后再慢慢扩展范围。6.3 Savepoint 恢复后数据重复或丢失这个坑在断点续传验证流程中非常常见。现象是创建 Savepoint 后取消了作业然后在 MySQL 里做了增删改操作再从 Savepoint 恢复作业看到日志里既有历史数据的重放又有恢复之后的新变更。原因在于 Flink CDC 的位点状态保存在状态中而状态中不仅包含 Binlog 位点还包含已经处理过的数据变化。如果 Savepoint 创建的时刻和取消作业的时刻之间存在时间差作业可能已经消费了更多 Binlog 事件但尚未触发新的 Checkpoint这些事件会丢失。反过来如果 Checkpoint 开启的间隔过短Savepoint 可能包含了部分未完全处理的数据恢复后引发重复消费。解决方案是在创建 Savepoint 之前先停止作业给 Flink 一个足够的缓冲时间来完成最后一次 Checkpoint然后再创建 Savepoint。此外需要确认状态后端是 HashMapStateBackend 还是 RocksDBStateBackend两者在恢复语义上有细微差别推荐生产环境使用 RocksDB 来降低大状态下的内存压力。6.4 Doris 写入时 Schema 类型不匹配导致任务失败最后一个高频问题是源端 MySQL 和目标端 Doris 的字段类型不一致。Doris 的 DECIMAL 类型在底层还是整型存储精度和标注不同例如 MySQL 的 DECIMAL(10,2) 同步到 Doris 的 DECIMAL(10,2) 一般没有问题但如果某个字段在 MySQL 是 VARCHAR(255)而 Doris 里对应的维度表已经建成了 VARCHAR(100)那么超长值写入时就会报错。另外轻量级 Schema Change 默认打开如果表结构频繁变更Doris 可能在处理多个并发 ALTER 时发生冲突。我的处理方式是第一次涉及时先手动在 Doris 建表验证字段类型映射关系跑通后再启用 pipeline 自动建表生产环境建议关闭 light_schema_change因为默认的同步一致性更可控。从这四条踩坑记录里能总结出一个共性Flink CDC 链路的大部分问题出在「源端配置」和「目标端兼容性」上而不是 Flink 计算层本身。花时间把 MySQL 侧 Binlog 配置和 Doris 建表规范提前梳理清楚后续的同步作业会省很多心力。7. 二次开发实战如何把这套同步体系改造为更适合你业务的方案7.1 增加字段过滤与敏感信息脱敏原始案例把整行数据原样同步到 Doris但真实业务往往不是这样。比如订单表里有用户手机号、身份证号你不想把它们原样落到数仓或者目标表只需要业务核心字段不需要 Binlog 的元数据字段。改造思路是在 Flink 层面做一层数据投影与脱敏。如果是 FlinkSQL 模式最简单的做法是把 SELECT 的列收窄CREATE TABLE mysql_cdc_orders ( id STRING PRIMARY KEY NOT ENFORCED, user_id STRING, phone STRING, order_amount DECIMAL(10,2), create_time TIMESTAMP(3) ) WITH ( connector mysql-cdc, hostname hadoop103, port 3306, username root, password 000000, database-name test, table-name t1 ); CREATE TABLE doris_orders_safe ( id STRING, user_id STRING, order_amount DECIMAL(10,2), create_time TIMESTAMP(3) ) WITH ( connector doris, fenodes node01:7030, table.identifier test.orders_safe, username root, password 000000, sink.properties.format json, sink.properties.read_json_by_line true ); -- 只同步业务字段 INSERT INTO doris_orders_safe SELECT id, user_id, order_amount, create_time FROM mysql_cdc_orders;如果你有更复杂的脱敏逻辑比如手机号中间四位用星号替代可以进一步用 SQL 的 REGEXP_REPLACE 实现或者在 DataStream 模式中自定义一个 map 算子在写 Doris 之前把敏感字段替换掉。这种做法的好处是改动范围仅在同步链路中间层源端和目标端完全无感知。7.2 整合进自研数据同步平台时的接口设计思路当你接入的 MySQL 实例越来越多光靠 YAML 文件管理会变得难以维护。常见的做法是用一个配置中心或元数据库来统一管理同步任务每个同步管道在业务侧通过一个抽象接口描述源端连接信息、目标端连接信息、表映射规则、同步模式(全量/增量/全量增量)、Schema 变更策略。Flink CDC 3.0 底层足够稳定之后上层平台做的其实就是把 YAML 变成模板、把模板变量存库让运维通过 Web 页面操作自动化生成同步任务。我一般会这样做把上面提到的 YAML 拆成「源配置」「目标配置」「路由配置」「Pipeline 配置」四个独立模板用 FreeMarker 或 Velocity 动态渲染再通过 REST API 提交到 Flink CDC。这样开发者在页面上配置好同步需求后系统自动生成模板随后触发同步任务。再配合一个监控模块定时扫描任务运行指标比手动改 YAML 文件的方式可维护性高不少。7.3 状态一致性与指标监控的强化Flink CDC 3.0 的同步任务可靠性强依赖 Checkpoint但默认的 HashMapStateBackend 在状态量大的情况下容易因为堆内存溢出而触发 Full GC。我从实践经验里的建议是生产环境务必换成 RocksDBStateBackend并且把 CheckpointStorage 选择在 HDFS 或独立的分布式文件系统上避免本机磁盘问题导致状态丢失。如果需要监控同步任务的延迟和吞吐指标可以在 Sink 端增加自定义回调函数统计每条记录从 Binlog 捕获到写入 Doris 的端到端耗时定期写入 Prometheus。一旦发现延迟超过业务阈值通过告警规则及时通知值班人员。这套体系在项目上线后给我省了不少半夜被叫起来处理问题的时间。从那以后我每次搭建新的同步管道都会强制走一遍源端配置验证、Checkpoint 参数调优、Schema 变更策略确认、监控告警接入。这四个环节缺一不可希望对你的项目也能有所帮助。本文还有配套的精品资源点击获取

Flink CDC 3.0实战：MySQL到Doris实时同步全攻略

相关推荐

ai去ai味技巧红黑榜：6种降AI率方法同文实测，论文AIGC检测只有1种从100%降到0%

会议纪要哪个软件安全性高？5款主流工具实测：从加密机制到权限管控全拆解

从软件到硬件：后端开发者的嵌入式入门踩坑手记

更多资讯

LangChain实战指南：从核心抽象到企业知识库问答落地

Lambda架构实战：批处理与流处理融合的实时数据平台设计

Spring Boot招聘数据可视化系统开发实战：从数据采集到可视化大屏

AI生成PPT效率提升300%：职场人必看的Aipy实操指南

d3dx9_42.dll丢失怎么办？官方DirectX运行库修复方法详解

PFC2D热力耦合岩石热损伤模拟：从颗粒流代码到参数标定

今日更新

悟空开源无代码平台实战：从实体建模到应用设计避坑指南

基于.NET9的多租户权限系统：数据权限与动态API落地实践

场地预约小程序毕设实战：SSM+微信小程序+MySQL数据库设计到避坑

本周精选

金融AI落地指南：从调研报告到技术选型与工程实践

大模型上下文成本与注意力优化技术深度解析——为何O(n²)平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测

基于GMM的语音识别系统原理与MATLAB实现详解

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？