Flink CDC 3.0实战：MySQL到Doris实时同步全流程拆解-北京尧图网络

简介这是一份面向有一定Flink基础的大数据研发人员的Flink CDC 3.0技术文档由尚硅谷研究院编写系统讲解CDC概念、基于查询与基于Binlog两种方式的差异并重点演示flink-cdc-connectors组件如何从MySQL等数据库读取全量与增量变更数据。文档包含完整案例实操覆盖DataStream与Flink SQL两种应用方式逐步指导完成MySQL到Doris的Streaming ETL同步涉及环境搭建、配置编写、任务启动与测试验证同时强调开启Binlog、设置检查点等关键细节。资源为单个docx文档大小145KB内容结构清晰配有代码示例和操作指南适合用于理解原理、上手实践及排查同步问题。该文档已有1788人浏览学习适合需要快速掌握实时数据同步能力的工程师参考。1. Flink CDC 3.0MySQL 到 Doris 实时同步的全流程拆解以前做实时数仓MySQL 到 Doris 的数据同步是最折磨人的环节定时任务轮询延迟高DataX 抽全量又给源库加压力业务侧一问数据怎么还没到只能干瞪眼。这套 Flink CDC 3.0 全流程讲义把「MySQL 数据实时捕获与同步至 Doris」这件事从原理到落地讲得很透——基于 Binlog 的变更捕获机制、3.0 版本主推的 YAML 管线式 StreamingETL、DataStream 和 FlinkSQL 两种接入方式一条命令就可以把整库变更流进 Doris还支持路由改名和断点续传。适合有 Flink 基础、正在搭实时数仓或做数据同步的工程师照着步骤能复现也能把底层逻辑看明白。2. CDC 原理与选型基于查询和基于 Binlog 的取舍2.1 两种 CDC 模式的本质差异CDCChange Data Capture的核心思想不复杂监测并捕获数据库的变动——包括数据行的插入、更新、删除也包括表结构的变化——按发生顺序完整记录下来写入消息中间件供其他服务订阅和消费。真正拉开差距的是「怎么捕获」这件事业界分成两条完全不同的路线。基于查询的 CDC 是轮询派周期性执行 SQL 查询把变化的数据捞出来。开源产品里 Sqoop、DataX 属于这一类。它的优点是实现简单对数据库版本没有要求缺点是只能查出「当前状态」中间被修改过的数据直接丢了每次全表扫描还会给源库造成不小的压力延迟也高。基于 Binlog 的 CDC 是日志派直接读 MySQL 的二进制日志依靠 binlog_formatrow 记录每一行变更前后的完整镜像。开源产品有 Canal、Maxwell、Debezium。这类方式能拿到完整的历史变更序列延迟低也不增加数据库查询压力。下面这张对照表把差异列得很清楚我做选型时基本就看这一张表对比项基于查询的 CDC基于 Binlog 的 CDC开源产品Sqoop、DataXCanal、Maxwell、Debezium执行模式BatchStreaming是否捕获所有数据变化否是延迟性高延迟低延迟是否增加数据库压力是否选型结论也很直接对延迟敏感、要拿全部增删改记录的场景直接走 Binlog 路线临时导数据、离线批处理用查询型工具反而更省事毕竟不用动数据库配置。2.2 flink-cdc-connectorsFlink 生态里的 CDC 组件Flink 社区在 CDC 这条路上做了一个很关键的组件flink-cdc-connectors。它是一个 source 组件可以直接从 MySQL、PostgreSQL 等数据库读取数据而且天然支持「全量增量」一起读——第一次启动做全量快照快照完成后无缝切换到 Binlog 增量读取用户侧完全不需要关心快照和增量之间的衔接问题。这个设计对使用者非常友好。以前做数据同步全量和增量是两个任务中间还有数据窗口期用 Flink CDC一个 source 就把两段串起来了。再配合 Flink 的 Checkpoint 机制Binlog 的读取位置会作为状态保存下来任务挂了可以从检查点续跑这就是后面要展开讲的断点续传。组件本身是开源的Maven 坐标是 com.ververica:flink-connector-mysql-cdc版本跟 Flink 版本有对应关系第 3 章会给出具体搭配。2.3 开启 Binlog让 MySQL 先「开口说话」Binlog 在 MySQL 8.0 里默认是开启的但很多自建实例或云上实例出于性能考虑会关掉。Flink CDC 要读变更第一步就是确认目标库的 Binlog 配置。讲义里的做法是在 /etc/my.cnf 里加一段配置# 数据库实例唯一 id复制拓扑中不能重复 server-id 1 # 开启 binlog该参数值会作为 binlog 文件名前缀 log-binmysql-bin # binlog 格式必须为 rowCDC 工具需要行级镜像 binlog_formatrow # 只记录指定库的变更需根据实际情况修改 binlog-do-dbtest binlog-do-dbtest_route四个参数各有用处server-id 是实例在复制拓扑里的唯一标识同一套环境里不能重复log-bin 开启后 MySQL 会在数据目录里滚动生成 mysql-bin.000001 这类文件binlog_formatrow 最关键它决定 Binlog 记录的是每行数据变更前后的完整镜像而不是 SQL 语句CDC 工具只有拿到 row 格式才能还原出 update 和 delete 的具体内容binlog-do-db 做库级过滤只记录 test 和 test_route 两个库的变更省磁盘也省同步开销。改完配置文件必须重启 MySQL 才生效。验证方式也不复杂登录 MySQL 执行 SHOW VARIABLES LIKE log_bin;看到 ON 说明已开启再执行 SHOW VARIABLES LIKE binlog_format;确认是 ROW。注意binlog-do-db 是个容易埋雷的参数以后新加库进来必须改配置再重启否则新库的变更不会进 Binlog。避坑章节我会再展开。3. DataStream 方式实操Checkpoint 与 Savepoint 断点续传3.1 Maven 依赖与打包配置DataStream 方式是最贴近 Flink 编程模型的写法把 CDC Source 当成一个普通的 DataStream Source用 env.fromSource 接进来后面接任意算子和 sink。讲义里的依赖配置如下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 !-- Flink DataStream 核心依赖 -- 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 !-- Table 相关依赖FlinkSQL 方式会复用同一套 -- 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 !-- CDC 核心MySQL Source -- dependency groupIdcom.ververica/groupId artifactIdflink-connector-mysql-cdc/artifactId version3.0.0/version /dependency !-- MySQL JDBC 驱动 -- 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-connector-mysql-cdc 的版本要跟 Flink 版本匹配讲义里 Flink 用 1.18.0、CDC connector 用 3.0.0这组搭配是验证过的。mysql-connector-java 用 8.0.31对应 MySQL 8.0 服务端。打包用 maven-assembly-plugin 生成 jar-with-dependencies 的 fat jar——任务要提交到 Flink 集群上跑集群节点不一定有这些依赖打成一个大包是最省心的做法。3.2 核心代码MySqlSource 构建与 Checkpoint 配置讲义里的 DataStream 示例代码有两大重点Checkpoint 配置和 MySqlSource 的构建参数。完整代码长这样import com.ververica.cdc.connectors.mysql.source.MySqlSource; import com.ververica.cdc.connectors.mysql.table.StartupOptions; import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.restartstrategy.RestartStrategies; import org.apache.flink.api.common.time.Time; import org.apache.flink.runtime.state.hashmap.HashMapStateBackend; import org.apache.flink.streaming.api.CheckpointingMode; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.CheckpointConfig; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; public class FlinkCDCDataStreamTest { public static void main(String[] args) throws Exception { // 1. 流处理环境 StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 2. 开启 CheckpointFlink-CDC 把 binlog 读取位置存进状态 // 想断点续传必须从 Checkpoint 或 Savepoint 启动 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); // 任务取消时保留最后一次 CK 数据方便手动恢复 env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 失败重启策略1 天内最多重启 3 次间隔 1 分钟 env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L))); // 状态后端HashMap HDFS 存储 env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage(hdfs://hadoop102:8020/flinkCDC); // 访问 HDFS 的用户名 System.setProperty(HADOOP_USER_NAME, atguigu); // 3. 构建 MySqlSource MySqlSourceString mySqlSource MySqlSource.Stringbuilder() .hostname(hadoop103) .port(3306) .databaseList(test) // 捕获的数据库 .tableList(test.t1) // 捕获的表 .username(root) .password(000000) .deserializer(new JsonDebeziumDeserializationSchema()) // SourceRecord 转 JSON String .startupOptions(StartupOptions.initial()) .build(); // 4. 接入 env打印输出 DataStreamSourceString mysqlDS env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), MysqlSource); mysqlDS.print(); env.execute(); } }代码里 Checkpoint 部分的注释值得细看。CDC Source 的断点续传依赖状态Flink 会把当前读到的 binlog 文件名和 offset 存进 Checkpoint任务失败后从最近一次成功的 Checkpoint 恢复就能接着上次的位置继续读不丢也不重。所以 env.enableCheckpointing(3000L, EXACTLY_ONCE) 不是锦上添花而是 CDC 任务的刚需setCheckpointTimeout 控制单次 Checkpoint 超时时间minPauseBetweenCheckpoints 防止 Checkpoint 过于频繁互相挤占enableExternalizedCheckpoints 设置取消时保留最后一次 CK是手动恢复的前提。MySqlSource 的构建参数里databaseList 和 tableList 支持库表过滤tableList 的格式是「库名.表名」deserializer 用了 JsonDebeziumDeserializationSchema把 Debezium 的 SourceRecord 转成 JSON 字符串print 出来就能看到完整的变更事件。startupOptions 决定任务启动时的读取位置五种模式差别很大启动模式行为说明initial首次启动先做全量快照再继续读最新 binlog默认推荐earliest不做快照从 binlog 最开头读需保证 binlog 保留完整历史latest不做快照只读启动后的增量启动前的数据拿不到specificOffset从指定的 binlog 文件偏移量开始读timestamp从指定时间戳开始读跳过更早的变更提示initial 是最省心的选择兼顾存量数据和增量数据latest 适合只关心启动后变化的场景但它不会帮你补历史数据很容易误用。3.3 提交任务与 Savepoint 恢复验证讲义里的测试流程是一条完整链路值得照着走一遍。先把项目打成 fat jar 传到集群节点然后依次启动 HDFS 和 Flink 集群# 启动 HDFS start-dfs.sh # 启动 Flink 集群 bin/start-cluster.sh # 提交 CDC 任务到集群 bin/flink run -m hadoop102:8081 -c com.atguigu.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar任务跑起来后观察 TaskManager 日志能看到它从 MySQL 的 test.t1 表把全量数据读出来。这时候做断点续传验证先给任务创建 Savepoint然后在 WebUI 上取消任务接着在 MySQL 里修改或删除数据最后从 Savepoint 重启# 创建 Savepoint bin/flink savepoint JobId hdfs://hadoop102:8020/flinkCDC/save # 从 Savepoint 恢复重启 bin/flink run -s hdfs://hadoop102:8020/flinkCDC/save/savepoint-xxxx \ -c com.atguigu.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar恢复后观察日志会发现程序从上次保存的 binlog 位置继续读取消期间产生的变更也能补上——这就是 CDC 任务最核心的可靠性保证。实战里我还会多做一步取消前记下 MySQL 当前数据行数恢复后对比读出来的行数和 binlog 事件数确认不丢才算验证通过。4. FlinkSQL 与 Pipeline 方式从 SQL DDL 到 YAML 配置4.1 FlinkSQL 方式mysql-cdc Connector 的建表 DDLDataStream 方式灵活但代码量大。如果只是想把 MySQL 表映射成 Flink 表做查询或同步FlinkSQL 方式显然更轻——不需要写 Source 构建逻辑一张 CREATE TABLE 的 DDL 就搞定了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); 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(); } }这段代码里有两个容易忽略的细节。第一个是 PRIMARY KEY NOT ENFORCEDFlink Table 不会强制主键约束但这个声明对 CDC 很重要——update 和 delete 事件要靠它定位要更新的行Source 内部会用主键做变更数据的合并。第二个是 WITH 里的 connector mysql-cdcFlinkSQL 通过这个标识自动找到 flink-connector-mysql-cdc 插件不需要像 DataStream 那样手动 new Source。hostname、port、database-name、table-name 这些参数跟 DataStream 的 builder 一一对应只是换成了 SQL 风格。FlinkSQL 方式的测试也更简单本地 IDE 直接跑 main 方法就能看到结果提交到集群的方式跟 DataStream 完全一样打成 fat jar 用 flink run 提交即可。4.2 MySQL 到 Doris 的 Streaming ETLYAML 配置拆解3.0 版本最大的变化是引入 Pipeline管线模式不用写 Java 代码一份 YAML 配置文件描述 source、sink、route、pipeline然后用 flink-cdc.sh 直接启动。讲义里 MySQL 到 Doris 的同步变更案例配置如下source: type: mysql hostname: node01 port: 3306 username: root password: 123 tables: test.\.* # 正则匹配 test 库下所有表 server-id: 5400-5404 # 一组 server-id供并行读取 binlog server-time-zone: UTC8 sink: type: doris fenodes: node01:7030 # Doris FE 地址 username: root password: 000000 table.create.properties.light_schema_change: true # 开启轻量 Schema 变更 table.create.properties.replication_num: 1 # 副本数测试环境设 1 pipeline: name: Sync MySQL Database to Doris parallelism: 1每个字段都不是随便写的。tables 用 test..* 这种正则表达式表示捕获 test 库下所有表比 DataStream 里写死 test.t1 灵活得多这也是 Pipeline 模式的主打能力——整库同步。server-id 配的是区间 5400-5404 而不是单个值因为 Pipeline 默认会有多个并行子任务去读 binlog每个子任务需要独立的 server-id配区间避免冲突。sink 部分两个 table.create.properties 是给 Doris 建表用的light_schema_change 让 Doris 支持轻量级 schema 变更加列、改列不需要重写数据文件replication_num 控制副本数测试环境设 1 可以省存储。启动流程讲义写得很细。先要在 Flink 的 conf/flink-conf.yaml 里打开 Checkpoint# 1. 修改 Flink 配置每隔 5 秒做一次 Checkpoint vim conf/flink-conf.yaml # 在文件中加入execution.checkpointing.interval: 5000 # 2. 分发配置到所有 Flink 节点后启动集群 bin/start-cluster.sh # 3. 启动 Doris FE 和 BE bin/start_fe.sh bin/start_be.sh # 4. 连接 Doris 并创建目标库 mysql -uroot -p000000 -P9030 -hhadoop102 create database test; # 5. 启动 Flink CDC 同步任务 bin/flink-cdc.sh job/mysql-to-doris.yaml注意flink-cdc.sh 会把任务提交到 Flink 集群启动前先确认 Flink 的 Checkpoint 配置已生效否则任务挂了没法自动恢复。任务起来后刷新 Doris 的 test 库能看到 MySQL test 库下的 t1、t2、t3 三张表已经在 Doris 里建好数据也同步过来了。这时候在 MySQL 侧做新增、修改、甚至新增列的操作再刷新 Doris会发现数据变化和表结构变化都跟着过来了。4.3 同步变更与路由变更两种管线的区别Pipeline 模式里「同步变更」和「路由变更」是两种常见场景讲义各给了一个案例。同步变更对应上面的配置源库 test 同步到 Doris 后库名不变、表名不变适合库名一致的场景。路由变更则要解决「库名或表名不一样」的问题比如 MySQL 的 test_route 库要落到 Doris 的 doris_test_route 库source: type: mysql hostname: hadoop103 port: 3306 username: root password: 000000 tables: test_route.\.* server-id: 5400-5404 server-time-zone: UTC8 sink: type: doris fenodes: node01:7030 username: root password: 000000 # 路由规则源表映射到目标表实现库名/表名改名 route: - source-table: test_route.t1 sink-table: doris_test_route.t1 - source-table: test_route.t2 sink-table: doris_test_route.t2 - source-table: test_route.t3 sink-table: doris_test_route.t3 pipeline: name: Route MySQL Database to Doris parallelism: 1route 段就是改名规则source-table 和 sink-table 一一对应。这种场景在真实业务里非常常见源库是业务库、表名混乱目标库要按数仓规范重新命名或者多个源库要汇聚到一个目标库路由规则可以解决同名冲突。验证方式跟同步变更一样在 MySQL 的 test_route 库里改数据、加列然后刷新 Doris 的 doris_test_route 库观察变化。5. 避坑与排查Binlog、Checkpoint、Doris 侧的常见坑这套讲义我前后跑了两遍第一遍照着敲完第二遍故意制造故障验证恢复能力。中间踩了不少坑挑五个最典型的记在这里按「现象 → 原因 → 解决」的顺序写遇到同类问题可以直接对照排查。5.1 Binlog 环节配置不生效与 server-id 冲突第一个坑改完 my.cnf 重启 MySQLFlink CDC 任务还是读不到增量变更。现象是任务能启动、全量数据也正常打印但在 MySQL 里 insert 数据后TaskManager 日志没有任何反应。原因排查下来有两个方向最常见的是 binlog_format 没生效MySQL 8.0 默认格式虽然是 ROW但很多从 5.7 升上来的实例还停在 STATEMENTCDC 工具拿不到行镜像另一个是 binlog-do-db 只配了 test 和 test_route而实际操作的表不在这些库里。解决方法是登录 MySQL 执行 SHOW VARIABLES LIKE binlog_format先确认是 ROW再确认 binlog-do-db 覆盖了目标库最后重启 mysqld 让它真正加载新配置。这三步按顺序做完九成问题能解。第二个坑多个 CDC 任务连同一个 MySQL跑一会儿某个任务就报错断开日志里出现 Got error reading packet from server。原因是这些任务用了相同的 server-id。MySQL 的 binlog dump 协议要求每个消费端有唯一 server-id重复 id 会被后来的连接挤掉表现就是先启动的任务突然断开。解决方法是给每个 CDC 任务分配独立的 server-id或者像讲义里那样配一个区间 5400-5404。注意区间大小要大于任务的实际并行度否则并行子任务之间还是会撞。5.2 Checkpoint 环节HDFS 权限与恢复路径第三个坑任务启动后报 HDFS 权限错误Checkpoint 写不进去任务反复重启。现象是日志里出现 Permission denied指向 hdfs://hadoop102:8020/flinkCDC 这个路径。原因是提交任务的用户和 HDFS 上的目录属主不一致。解决方法是代码里加 System.setProperty(HADOOP_USER_NAME, atguigu)并且保证这个路径在 HDFS 上已创建。我一般把这行放在 main 方法最前面避免任务跑起来才报错另外如果不用 HDFS 而是本地文件系统要确认所有 TaskManager 节点都能访问同一个路径否则状态后端不一致也会翻车。第四个坑从 Savepoint 恢复后读取位置不对要么重复消费一段要么丢了取消期间的变更。现象是恢复后日志里打印的 binlog offset 和取消前对不上。原因多半是 -s 参数指定错了目录——savepoint 目录下会有多个 savepoint-xxxx 子目录每个对应一次手动 savepoint命令行里手敲路径很容易指错。解决方法是创建 savepoint 后把命令输出的完整路径复制下来恢复时原样粘贴不要自己拼。另外恢复用的 jar 包必须和创建 savepoint 时是同一份代码类结构变了状态恢复会直接失败。5.3 Doris 环节表结构变更不生效第五个坑MySQL 侧 ALTER TABLE 加了一列Doris 里的表结构没跟着变新列的数据也没进来。现象是 Pipeline 任务运行正常Doris 查询却看不到新列。原因是 sink 配置里没开启 light_schema_changeDoris 默认的 schema change 流程是重写数据文件在 CDC 实时 DDL 场景下往往不会自动触发。解决方法是确认 sink 配置里 table.create.properties.light_schema_change: true同时确认 Doris 版本支持轻量 schema 变更。如果配置没问题还是不生效先手动在 Doris 执行 ALTER TABLE 补上列让后续增量能正常写入再来排查 Pipeline 侧 DDL 事件是不是被过滤掉了。6. 二次开发从单表同步到多库多表动态路由讲义里的案例停留在单表捕获、固定路由实际业务里需求会往前走好几步。我基于这套代码做过一轮改造几个方向可以直接复用。第一把 DataStream 方式里写死的 databaseList(test) 改成可配置。做法是把库表列表、用户名密码、启动模式全部抽到 properties 文件或启动参数里main 方法只做组装。这样同一个 jar 包换一套参数就能跑不同数据源不用每次改代码重新打包。第二自定义 DeserializationSchema。JsonDebeziumDeserializationSchema 输出的 JSON 里带 schema 描述下游 Kafka 或 Doris sink 不一定需要。常见做法是实现 DebeziumDeserializationSchema 接口把 SourceRecord 解析成业务字段的 POJO顺便把 op 字段c/u/d 对应增删改单独拎出来方便下游做分流处理。第三Pipeline 模式下把正则表匹配和路由规则做成动态的。tables 用 test..* 已经能覆盖整库但路由规则如果新增表YAML 里要手动加一条。我改造时写了个脚本扫描 MySQL 里的表清单自动生成 route 段再调用 flink-cdc.sh 重启任务整个过程不用手改配置。验证方法也固定成了一套先核对 Binlog 参数再起任务看全量然后手动做一次 insert、一次 update、一次 delete、一次 add column最后从 Savepoint 恢复一次确认位点正确。这套验证做完任务上线基本不会翻车。从那以后我每次上线 CDC 任务都强制走一遍这套流程哪怕只是改一个表名。这套讲义把从 CDC 原理到 Pipeline 落地的完整链路都覆盖了代码和 YAML 配置都可以直接抄希望帮到你。本文还有配套的精品资源点击获取

Flink CDC 3.0实战：MySQL到Doris实时同步全流程拆解

相关推荐

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 成本测算与选型避坑（附配置）

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