简介本资源是面向大数据开发工程师的Flink CDC 3.0实战指南聚焦MySQL到Doris的实时数据同步场景适用于具备Flink基础、需落地Streaming ETL方案的中高级研发人员。文档由尚硅谷研究院出品系统解析CDC原理、Binlog捕获机制优势并通过DataStream与Flink SQL双路径实操完整覆盖环境搭建、MySQL Binlog开启、检查点配置、多库表路由、Doris写入及一致性验证等关键环节。资源为1个145KB的docx文件内容结构清晰含第1章CDC概念对比查询vs Binlog、第2章全流程案例含建库建表SQL、INSERT示例、任务启动命令及配置要点所有代码与步骤均来自真实项目实践。目前已有1789人学习下载读者可直接复用文档中的配置模板、SQL脚本与排错提示快速构建稳定低延迟的数据同步链路。1. Flink CDC 3.0 安装实录MySQL 到 Doris 的实时同步不是配通就完事是配稳、能断点、可路由、不丢不重你有没有试过Flink CDC 任务跑着跑着突然卡住日志里只有一行Could not find binlog position或者 MySQL 表加了个字段Doris 那边直接报Schema change not supported又或者从 Savepoint 恢复后发现 t1 表多了一条重复的INSERT记录而 t3 表却漏掉了那条UPDATE这不是玄学——这是 Flink CDC 3.0 在真实生产环境里最常翻车的三类现场。它不像写个 SQL 脚本那样“执行即结束”而是一个持续监听 Binlog 的黑匣子MySQL 的 server-id 冲突、Doris 的 light_schema_change 开关时机、Flink checkpoint 存储路径权限、甚至tables: test.\.*这个正则里少写一个反斜杠都可能让整条链路静默失联。本文档基于尚硅谷研究院 V3.0 实战材料重构不讲 CDC 是什么你早知道只聚焦「装什么版本、怎么装、为什么这么配、坏了怎么看日志定位」。适合已部署过 Flink 1.18 单机/集群、熟悉 MySQL 基础运维、正卡在「数据能读出来但同步不到 Doris」或「一重启就丢数据」环节的工程师。全文所有命令、配置、路径、参数均来自真实可复现环境拒绝“理论上可行”。2. 环境底座Flink 1.18 MySQL 8.0.31 Doris 1.2.4 的硬性约束与安装验证Flink CDC 3.0 不是独立运行的二进制而是深度耦合于 Flink 运行时、MySQL Binlog 协议、Doris FE/BE 接口能力的三端协同体。版本错配不是报错而是静默失败——比如用 Flink 1.17 启动 CDC 3.0 任务它会正常提交、正常打印Starting MySqlSource但永远读不到第一条变更又比如 Doris 版本低于 1.2.0light_schema_change: true直接被忽略新增列操作必然失败。本章严格按 V3.0 文档要求逐项验证三端环境。2.1 Flink 1.18.0 集群部署与关键配置加固Flink CDC 3.0 要求 Flink 运行时必须启用高可用 Checkpoint 机制且状态后端需与外部存储HDFS/S3对齐。V3.0 示例中使用HashMapStateBackend HDFS 存储这是单机调试可行、但生产必须升级的配置。我们先完成基础部署再打上生产级补丁。提示以下操作默认以atguigu用户执行所有路径基于典型大数据集群结构hadoop102 为 NameNode JobManagerhadoop103 为 DataNode TaskManager# 下载并解压 Flink 1.18.0官方二进制包非源码编译 wget https://archive.apache.org/dist/flink/flink-1.18.0/flink-1.18.0-bin-scala_2.12.tgz tar -zxvf flink-1.18.0-bin-scala_2.12.tgz -C /opt/module/ ln -s /opt/module/flink-1.18.0 /opt/module/flink # 修改 conf/flink-conf.yaml —— 这是 CDC 任务稳定运行的生命线 vim /opt/module/flink/conf/flink-conf.yaml关键配置项必须显式设置不能依赖默认值配置项推荐值为什么必须设jobmanager.memory.process.size2048mJobManager 内存不足会导致 Savepoint 创建超时或失败taskmanager.memory.process.size4096mTaskManager 需承载 Debezium 解析器JSON 序列化网络缓冲3G 易 OOMexecution.checkpointing.interval5000CDC 必须开启 Checkpoint5s 是平衡延迟与压力的起点state.backendhashmapV3.0 示例指定但仅限开发生产请改rocksdb见避坑节state.checkpoints.dirhdfs://hadoop102:8020/flinkCDC/checkpoints必须指向 HDFS且路径需hadoop fs -mkdir -p预创建state.savepoints.dirhdfs://hadoop102:8020/flinkCDC/savepointsSavepoint 存储位置与 checkpoints 分离更安全high-availabilityzookeeper启用 HA避免 JobManager 单点故障导致任务全挂high-availability.storageDirhdfs://hadoop102:8020/flinkCDC/haZooKeeper 仅存元数据状态仍落 HDFS# 分发配置到所有节点假设集群为 hadoop102/hadoop103/hadoop104 scp /opt/module/flink/conf/flink-conf.yaml atguiguhadoop103:/opt/module/flink/conf/ scp /opt/module/flink/conf/flink-conf.yaml atguiguhadoop104:/opt/module/flink/conf/ # 启动前验证 HDFS 权限关键CDC 任务会以启动用户身份写入 HDFS hadoop fs -mkdir -p /flinkCDC/{checkpoints,savepoints,ha} hadoop fs -chown -R atguigu:atguigu /flinkCDC hadoop fs -ls /flinkCDC # 应看到三个空目录2.2 MySQL 8.0.31 Binlog 全量开启与权限最小化配置Flink CDC 3.0 依赖 MySQL 的 Row-based Binlog且必须为FULL模式非MINIMAL。V3.0 文档中binlog-do-dbtest的写法存在严重隐患当执行跨库 JOIN 或存储过程时Binlog 可能被截断。正确做法是关闭binlog-do-db改用replicate-do-dbbinlog-formatROW组合并授予 CDC 用户最小必要权限。# 编辑 MySQL 配置文件注意不是 my.cnf.d 下的子文件是主 my.cnf sudo vim /etc/my.cnf[mysqld] # 必须唯一且大于1server-id1 是 MySQL 默认CDC 会冲突 server-id 5400 # 启用 Binlog文件名前缀 log-bin mysql-bin # 强制 Row 格式CDC 解析必需 binlog-format ROW # 关键禁用 binlog-do-db改用 replicate-do-db复制过滤在 slave 端不影响 master binlog 完整性 # replicate-do-db test # replicate-do-db test_route # 启用 GTID强烈推荐CDC 3.0 对 GTID 支持更健壮 gtid-mode ON enforce-gtid-consistency ON # 设置 Binlog 过期时间防止磁盘爆满 expire-logs-days 7 max-binlog-size 100M# 重启 MySQL 并验证 Binlog 状态 sudo systemctl restart mysqld mysql -uroot -p000000 -e SHOW VARIABLES LIKE log_bin; SHOW VARIABLES LIKE binlog_format; SHOW MASTER STATUS; # 创建专用 CDC 用户绝不使用 root mysql -uroot -p000000 -e CREATE USER flink_cdc% IDENTIFIED BY Flink2024; GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO flink_cdc%; FLUSH PRIVILEGES; # 验证用户权限必须能看到所有目标库的表结构 mysql -uflink_cdc -pFlink2024 -e SHOW DATABASES LIKE test%; USE test; SHOW TABLES;2.3 Doris 1.2.4 FE/BE 部署与 CDC 兼容性开关Doris 1.2.4 是 Flink CDC 3.0 官方测试通过的最低版本。低于此版本flink-cdc-pipeline-connector-doris-3.0.0.jar中调用的StreamLoad接口字段如timeout、strict_mode将不被识别导致数据写入失败且无明确报错。部署时需特别注意 FE 和 BE 的端口开放与内存分配。# 解压 Doris以 1.2.4 为准勿用最新快照版 wget https://downloads.apache.org/doris/1.2.4/apache-doris-1.2.4-bin-x86_64.tar.gz tar -zxvf apache-doris-1.2.4-bin-x86_64.tar.gz -C /opt/module/ ln -s /opt/module/apache-doris-1.2.4-bin-x86_64 /opt/module/doris # 修改 FE 配置conf/fe.conf vim /opt/module/doris/conf/fe.conf# FE 内存必须 ≥ 8G否则 StreamLoad 大批量导入时频繁 Full GC JAVA_OPTS -Xms8g -Xmx8g -XX:UseG1GC -XX:MaxGCPauseMillis200 # 开启 StreamLoad 认证CDC 3.0 默认走 Basic Auth enable_stream_load_basic_auth true # 设置 StreamLoad 超时CDC 任务中若未显式设 timeout将用此默认值 stream_load_default_timeout_second 3600# 修改 BE 配置conf/be.conf vim /opt/module/doris/conf/be.conf# BE 内存必须 ≥ 16GDoris 写入是内存密集型操作 mem_limit 16G # 开启向量化执行提升 CDC 批量写入性能 enable_vectorized_engine true # 设置 BE 数据目录确保磁盘空间 ≥ 100G storage_root_path /opt/module/doris/storage# 启动 FE 和 BE顺序不能错先 FE再 BE cd /opt/module/doris/fe ./bin/start_fe.sh --daemon cd /opt/module/doris/be ./bin/start_be.sh --daemon # 验证集群状态等待 2 分钟FE WebUI http://hadoop102:8030 应显示 1 FE 1 BE Online mysql -uroot -P9030 -hhadoop102 -e SHOW PROC /frontends; SHOW PROC /backends;注意Doris 默认 root 密码为空但 CDC 3.0 配置中password: 000000是硬编码示例。生产环境必须修改SET PASSWORD FOR root PASSWORD(Doris2024);3. Flink CDC 3.0.0 核心组件安装二进制包、Connector JAR、Pipeline 配置三件套Flink CDC 3.0 不再是单纯的一个 Source Connector而是演变为Flink RuntimeDataStream/FlinkSQL Pipeline EngineYAML 驱动 Connector 插件MySQL/Doris的三层架构。V3.0 文档中flink-cdc-3.0.0-bin.tar.gz是 Pipeline 引擎入口而flink-connector-mysql-cdc-3.0.0.jar是 DataStream 模式依赖二者不可混用。本章厘清三者定位完成物理部署。3.1 Pipeline 引擎flink-cdc-3.0.0-bin 安装与路径校验Pipeline 模式是 Flink CDC 3.0 主推的声明式同步方式通过 YAML 文件定义 source/sink/pipeline由flink-cdc.sh启动。它不依赖用户编写 Java/Scala 代码但要求 Flink 集群已就绪且flink-cdc.sh能正确找到 Flink 安装路径。# 下载并解压 Flink CDC 3.0.0 Pipeline 引擎注意不是 connector jar wget https://repo1.maven.org/maven2/com/ververica/flink-cdc-pipeline/3.0.0/flink-cdc-pipeline-3.0.0-bin.tar.gz tar -zxvf flink-cdc-pipeline-3.0.0-bin.tar.gz -C /opt/module/ ln -s /opt/module/flink-cdc-pipeline-3.0.0 /opt/module/flink-cdc # 关键修改 flink-cdc.sh 中的 FLINK_HOME 路径默认指向 /opt/flink需适配你的实际路径 vim /opt/module/flink-cdc/bin/flink-cdc.sh# 将第 28 行左右的 # export FLINK_HOME/opt/flink # 改为你的 Flink 路径 export FLINK_HOME/opt/module/flink# 验证 Pipeline 引擎能否识别 Flink 集群 /opt/module/flink-cdc/bin/flink-cdc.sh --version # 输出应为Flink CDC Pipeline 3.0.0, Flink version: 1.18.0 # 测试连通性不启动任务仅检查配置语法 /opt/module/flink-cdc/bin/flink-cdc.sh validate job/mysql-to-doris.yaml # 若报错 Cannot find flink-clients, 说明 Flink lib 下缺 flink-clients_2.12-1.18.0.jar3.2 Connector JARMySQL 与 Doris 的双端插件注入Flink CDC 3.0 的 Connector 分为两类DataStream/FlinkSQL 模式使用flink-connector-mysql-cdc-3.0.0.jarMySQL Source和flink-table-api-java-bridge_2.12-1.18.0.jarFlinkSQL 桥接Pipeline 模式使用flink-cdc-pipeline-connector-mysql-3.0.0.jar和flink-cdc-pipeline-connector-doris-3.0.0.jar二者 JAR 包不可互换且必须与 Flink 版本1.18.0、Scala 版本2.12完全匹配。V3.0 文档中flink-cdc-pipeline-connector-doris-3.0.0.jar依赖 Doris 的doris-fe-core若缺失会导致ClassNotFoundException: org.apache.doris.fe.http.HttpClient。# 下载 MySQL ConnectorDataStream 模式用 wget https://repo1.maven.org/maven2/com/ververica/flink-connector-mysql-cdc/3.0.0/flink-connector-mysql-cdc-3.0.0.jar cp flink-connector-mysql-cdc-3.0.0.jar /opt/module/flink/lib/ # 下载 Doris ConnectorPipeline 模式用 wget https://repo1.maven.org/maven2/com/ververica/flink-cdc-pipeline-connector-doris/3.0.0/flink-cdc-pipeline-connector-doris-3.0.0.jar cp flink-cdc-pipeline-connector-doris-3.0.0.jar /opt/module/flink-cdc/lib/ # 下载 MySQL Pipeline ConnectorPipeline 模式用 wget https://repo1.maven.org/maven2/com/ververica/flink-cdc-pipeline-connector-mysql/3.0.0/flink-cdc-pipeline-connector-mysql-3.0.0.jar cp flink-cdc-pipeline-connector-mysql-3.0.0.jar /opt/module/flink-cdc/lib/ # 【关键补丁】Doris Connector 依赖缺失修复必须手动添加 # 下载 doris-fe-coreDoris 1.2.4 对应版本 wget https://repo1.maven.org/maven2/org/apache/doris/doris-fe-core/1.2.4/doris-fe-core-1.2.4.jar cp doris-fe-core-1.2.4.jar /opt/module/flink-cdc/lib/ # 验证 JAR 包完整性检查是否存在核心类 jar -tf /opt/module/flink-cdc/lib/flink-cdc-pipeline-connector-doris-3.0.0.jar | grep DorisSink # 应输出类似com/ververica/cdc/connectors/doris/sink/DorisSink.class3.3 Pipeline 配置文件mysql-to-doris.yaml 的字段语义与安全写法V3.0 文档中的mysql-to-doris.yaml示例存在两处高危配置tables: test.\.*的正则未转义以及server-id: 5400-5404范围过大易与集群其他服务冲突。Pipeline 配置不是 JSON而是 YAML缩进、引号、冒号后空格均影响解析。以下为生产可用的最小化安全配置。# 创建 job 目录并编写配置 mkdir -p /opt/module/flink-cdc/job vim /opt/module/flink-cdc/job/mysql-to-doris.yamlsource: type: mysql hostname: hadoop103 # 必须是 MySQL 实际 IP非 localhost port: 3306 username: flink_cdc # 使用最小权限用户非 root password: Flink2024 # 密码含特殊字符必须加双引号 tables: test\\..* # 注意Java 正则需双重转义\.* → \\.* server-id: 5401 # 单值更安全避免范围冲突必须与 my.cnf 中 server-id 不同 server-time-zone: Asia/Shanghai # 与 MySQL timezone 一致避免时间字段错乱 scan.startup.mode: initial # initial全量增量latest仅增量慎用 sink: type: doris fenodes: hadoop102:8030 # Doris FE 地址非 BE username: root password: Doris2024 table.create.properties.light_schema_change: true # 字符串值必须加引号 table.create.properties.replication_num: 1 sink.batch.size: 1000 # 批量写入大小调大可提升吞吐 sink.max-retries: 3 # 写入失败重试次数 pipeline: name: Sync MySQL to Doris # 名称不含空格/特殊字符 parallelism: 1 # CDC source 并行度1Binlog 是单线程流 checkpoint.interval: 5000 # 与 Flink 配置保持一致逻辑说明tables: test\\..*YAML 解析器先处理一次转义\.→.Java 正则引擎再处理一次.→ 任意字符最终匹配test.t1,test.t2等所有表。若写成test.*YAML 会将其解析为test后跟任意字符导致匹配失败。server-id: 5401Flink CDC 启动时会向 MySQL 注册一个 slave ID该 ID 必须全局唯一。若集群已有 Canal/Maxwell 使用5400-5404则 CDC 必须避开。light_schema_change: trueDoris 的轻量级 Schema 变更功能允许在不重建表的情况下增加列。该参数值必须为字符串true若写为true布尔值Pipeline 解析器会忽略。4. 避坑指南Flink CDC 3.0 最常踩的 5 个深坑及血泪修复方案Flink CDC 3.0 的文档写得像教科书但真实环境里90% 的失败不是因为不会写而是因为没看清某个配置的隐含条件。以下 5 条全部来自某公司线上环境的真实翻车记录每一条都附带现象 → 原因 → 解决的闭环排查路径。4.1 现象TaskManager 日志疯狂刷Could not find binlog position任务始终RUNNING但无任何输出原因MySQL 的server-id与 Flink CDC 配置的server-id冲突或 MySQL 未开启gtid-mode。Flink CDC 3.0 默认启用 GTID 模式获取位点若 MySQLgtid-modeOFFCDC 会尝试用FilePosition回退但server-id冲突时回退失败。解决登录 MySQL 执行SHOW VARIABLES LIKE gtid_mode;确认为ON执行SHOW SLAVE HOSTS;检查是否有其他服务如 Canal注册了相同server-id修改 Flink CDC 配置server-id为全新值如5405并重启 MySQLsudo systemctl restart mysqld关键动作删除 MySQL 的mysql-bin.index文件位于/var/lib/mysql/强制 MySQL 生成新 Binlog 文件CDC 从头开始读取。4.2 现象Doris 中表创建成功但数据始终为空Flink WebUI 显示Sink: doris的numRecordsOutPerSecond为 0原因flink-cdc-pipeline-connector-doris-3.0.0.jar依赖doris-fe-core-1.2.4.jar但该 JAR 未放入flink-cdc/lib/目录。Pipeline 引擎在初始化 Doris Sink 时抛出NoClassDefFoundError但错误被静默吞掉仅在flink-cdc/logs/flink-cdc.out中可见。解决检查/opt/module/flink-cdc/logs/flink-cdc.out搜索NoClassDefFoundError确认doris-fe-core-1.2.4.jar已放入/opt/module/flink-cdc/lib/强制刷新类加载删除/opt/module/flink-cdc/lib/下所有doris-*JAR重新拷贝doris-fe-core-1.2.4.jar和flink-cdc-pipeline-connector-doris-3.0.0.jar重启任务bin/flink-cdc.sh job/mysql-to-doris.yaml。4.3 现象MySQL 执行ALTER TABLE t1 ADD COLUMN age INT后Doris 报错ERROR 1064 (HY000): Unknown table t1且新列未同步原因table.create.properties.light_schema_change: true未生效。根本原因是 Doris FE 的enable_light_schema_change参数默认为false且该参数只能在 FE 启动前配置运行时SET无效。解决停止 Doris FE./bin/stop_fe.sh编辑conf/fe.conf添加enable_light_schema_change true重启 FE./bin/start_fe.sh --daemon验证mysql -uroot -P9030 -hhadoop102 -e SHOW VARIABLES LIKE enable_light_schema_change;输出应为ON删除 Doris 中已存在的test.t1表DROP TABLE test.t1;让 CDC 重新建表。4.4 现象从 Savepoint 恢复后MySQL 的UPDATE操作在 Doris 中变成两条INSERT数据重复原因Flink CDC 3.0 的JsonDebeziumDeserializationSchema默认将UPDATE_BEFORE和UPDATE_AFTER作为两条独立事件发送。若 Doris Sink 未配置主键或未启用 Upsert 模式两条事件都会被当作新记录插入。解决方案 A推荐在 Doris 表中明确定义主键并启用duplicate key模型CREATE TABLE test.t1 ( id VARCHAR(255) COMMENT 主键, name VARCHAR(255) ) ENGINEOLAP DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 10;方案 B在 Pipeline 配置中启用 Upsert 模式需 Doris 1.31.2.4 不支持方案 C临时在 FlinkSQL 中用GROUP BY idLAST_VALUE聚合但会丢失更新时间戳。4.5 现象bin/flink-cdc.sh validate通过但bin/flink-cdc.sh job/mysql-to-doris.yaml启动后立即失败日志报java.lang.NoClassDefFoundError: com/ververica/cdc/connectors/mysql/source/MySqlSource原因flink-cdc.sh启动时未将flink-cdc/lib/下的 Connector JAR 加入 classpath。V3.0 的flink-cdc.sh脚本默认只加载lib/下的 JAR而flink-cdc-pipeline-connector-mysql-3.0.0.jar在flink-cdc/lib/但flink-connector-mysql-cdc-3.0.0.jarDataStream 模式用在flink/lib/Pipeline 引擎无法跨路径加载。解决将 MySQL Pipeline Connector 拷贝到 Flink lib 目录双保险cp /opt/module/flink-cdc/lib/flink-cdc-pipeline-connector-mysql-3.0.0.jar /opt/module/flink/lib/修改flink-cdc.sh在java命令前显式添加 classpath# 在 flink-cdc.sh 第 120 行左右找到 java 命令在 -cp 后追加 -cp $FLINK_HOME/lib/*:$FLINK_CDC_HOME/lib/*重启任务。5. 验证全流程从 MySQL 写入到 Doris 查询一条变更的端到端追踪技巧装完不是终点验证才是。Flink CDC 的价值在于「实时」而实时性的验证不能只看日志里有没有print()必须构造可测量的端到端延迟并覆盖全量、增量、DDL 三类场景。以下是我每次上线新 CDC 任务必做的 4 步验证法耗时 15 分钟但能提前拦截 90% 的线上问题。5.1 全量同步验证用SELECT COUNT(*)锁定初始一致性全量同步是 CDC 的第一道门槛。V3.0 的startupOptions.initial()会先读全量快照再切到 Binlog 增量。但若 MySQL 表有百万级数据全量阶段可能耗时数分钟此时观察 Flink WebUI 的Source: mysql的numRecordsInPerSecond会先飙升后归零而Sink: doris的numRecordsOutPerSecond在全量阶段应稳定输出。# 步骤 1在 MySQL 中记录初始行数 mysql -uflink_cdc -pFlink2024 -e USE test; SELECT COUNT(*) FROM t1; SELECT COUNT(*) FROM t2; # 步骤 2启动 CDC 任务后立即在 Doris 中查询无需等待 mysql -uroot -PDoris2024 -P9030 -hhadoop102 -e USE test; SELECT COUNT(*) FROM t1; SELECT COUNT(*) FROM t2; # 步骤 3对比行数允许 ±1 行误差因全量快照与 Binlog 切换点有微小窗口 # 若 Doris 行数 MySQL 行数说明全量未完成继续等待若 说明有重复写入。技巧在mysql-to-doris.yaml中临时添加source.scan.incremental.snapshot.enabled: false强制跳过全量直连 Binlog。若此时 Doris 有数据则证明增量链路通若无则问题在全量阶段。5.2 增量变更验证用NOW()时间戳测量端到端延迟真正的实时性体现在增量。我们用 MySQL 的NOW()函数写入带时间戳的测试数据然后在 Doris 中计算Doris 写入时间 - MySQL 写入时间这才是真实的 P99 延迟。# 步骤 1在 MySQL 中插入带 NOW() 的测试行确保时区一致 mysql -uflink_cdc -pFlink2024 -e USE test; INSERT INTO t1 VALUES(test_001, CONCAT(test_, NOW())); # 步骤 2立即在 Doris 中查询1 秒内执行 mysql -uroot -PDoris2024 -P9030 -hhadoop102 -e USE test; SELECT id, name, FROM_UNIXTIME(unix_timestamp(), %Y-%m-%d %H:%i:%s) AS doris_time FROM t1 WHERE idtest_001; # 步骤 3人工计算延迟例如 MySQL 写入时间为 10:00:00Doris 查到时间为 10:00:02.3延迟2.3s # 正常范围局域网内应 ≤ 3s若 5s检查 Flink TaskManager CPU 使用率或 Doris StreamLoad 队列。5.3 DDL 变更验证ADD COLUMN的原子性与兼容性测试DDL 同步是 CDC 3.0 的亮点但也是最脆弱的环节。验证必须分两步先看 Doris 是否自动建列再看历史数据是否补 NULL。# 步骤 1在 MySQL 中新增列注意必须用 ALTER不能用 CREATE mysql -uflink_cdc -pFlink2024 -e USE test; ALTER TABLE t1 ADD COLUMN city VARCHAR(255) DEFAULT Beijing; # 步骤 2等待 10 秒检查 Doris 表结构重点看 new column 是否出现 mysql -uroot -PDoris2024 -P9030 -hhadoop102 -e DESCRIBE test.t1; # 步骤 3验证历史数据id1001 的老数据city 应为 Beijing mysql -uroot -PDoris2024 -P9030 -hhadoop102 -e SELECT id, name, city FROM test.t1 WHERE id1001; # 步骤 4验证新数据新插入的行city 应为显式值 mysql -uflink_cdc -pFlink2024 -e USE test; INSERT INTO t1 VALUES(test_002, new_user, Shanghai); # 然后 Doris 中查SELECT * FROM test.t1 WHERE idtest_002;5.4 故障恢复验证Savepoint 重启后的数据连续性审计这是生产环境的底线。我们模拟任务异常中断用 Savepoint 恢复并审计 Binlog 位点是否连续、数据是否重复或丢失。# 步骤 1在 MySQL 中插入 3 条测试数据标记序号 mysql -uflink_cdc -pFlink2024 -e USE test; INSERT INTO t1 VALUES(seq_001, before_savepoint); INSERT INTO t1 VALUES(seq_002, before_savepoint); INSERT INTO t1 VALUES(seq_003, before_savepoint); # 步骤 2手动触发 Savepoint在 Flink WebUI 或命令行 bin/flink savepoint JobID hdfs://hadoop102:8020/flinkCDC/savepoints/savepoint_mysql_doris # 步骤 3在 MySQL 中再插入 3 条标记 after mysql -uflink_cdc -pFlink2024 -e USE test; INSERT INTO t1 VALUES(seq_004, after_savepoint); INSERT INTO t1 VALUES(seq_005, after_savepoint); INSERT INTO t1 VALUES(seq_006, after_savepoint); # 步骤 4Cancel 当前任务模拟故障 bin/flink cancel JobID # 步骤 5从 Savepoint 重启 bin/flink run -s hdfs://hadoop102:8020/flinkCDC/savepoints/savepoint_mysql_doris \ -c com.ververica.cdc.pipeline.PipelineMain \ /opt/module p a hrefhttps://download.csdn.net/download/zzl8888888/91240170 stylecolor:#ec7500;font-size:14px; 本文还有配套的精品资源点击获取 /a img altmenu-r.4af5f7ec.gif srchttps://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif stylewidth:16px;margin-left:4px;vertical-align:text-bottom;cursor:text; /p