1. 先把话说明白多库多表同步到底在解决什么问题我在做实时数仓日常维护的时候最多被问到的一个需求就是“Doris 怎么同步多库多表”。说清楚一点就是上游有几十张甚至几百张表分散在多个业务库、多个业务系统里现在要全部实时搬到 Doris 里供 BI 报表、即席查询和数据服务使用。举个例子。某公司有十几套订单业务库每个库下面都有类似order_info、order_item、payment_record这种表库名字叫order_service_01、order_service_02……每套库的字段还不太一样有的多一个sale_channel有的多一个promotion_id。如果靠人工写导入脚本一张表一张表处理不仅工作量大而且线上微调一个字段就要改一遍流程。更麻烦的是业务还在不断拆库拆表今天加一张分表明天加一个字段完全跟不上节奏。“多库多表”这三个字拆开看是两件事第一数据分散在多个物理库中需要收集到同一个目标端第二每个库下面可能有多张表这些表要么是结构完全一致的同名表要么是结构近似的业务表要么是分库分表之后切出来的逻辑表。同步方案能不能落地关键就看能不能同时解决“库的扩展”和“表的规则识别”这两个问题。1.1 我见过的最典型场景先说三个我实际接触过的场景你对号入座就知道自己属于哪一种。第一种分库分表聚合。交易系统按用户 ID 或商户 ID 分库比如trade_db_0、trade_db_1每个库里都有一张trade_order表。Doris 这边要建一张大表dwd_trade_order_all把所有分片的数据合并进来并且保持主键唯一、实时更新。这是最常见也最考验方案设计的一种。第二种多系统数据汇聚。用户中心、订单中心、库存中心各自有独立的数据库每个库里几十张表要把这些 ODS 层的贴源数据同步到 Doris。这种场景不太需要关心业务逻辑合并但表数量非常多手工配置不现实需要整库同步能力。第三种多租户同构数据库合并。一套 SaaS 产品为每个租户单独开库库名带租户标识表结构基本一致偶尔有差异。要同步到 Doris 做全租户的运营分析这要求同步工具能通过正则匹配库名自动识别新增租户库。1.2 为什么不能靠传统离线工具硬扛很多人第一反应是用 DataX 或手工导 CSV 不行吗我直接说结论单次全量可以持续增量不行多库多表动态扩展更不行。离线批量同步的痛点是全量扫描对源库压力非常大。业务库白天要扛在线交易你每隔十分钟跑一次全量去扫几十张表慢的一批不说数据库 IO 和连接数先被拖垮。更重要的是BI 要看的是分钟级甚至秒级的实时数据离线 T1 或者十分钟一次根本无法覆盖。Doris 本身不是一个数据抓取工具它是一个高性能的分析型数据库。它提供了 Stream Load、Broker Load、Routine Load 这些导入通道但是“从源端数据库里把变化数据抓出来”这件事必须搭配一套 CDCChange Data Capture能力。所谓 CDC就是监听上游数据库的 binlog 或者 WAL 日志把插入、更新、删除操作实时解析出来再吐给下游。理解了这一点你就明白多库多表同步的核心其实是“CDC 工具 目标端导入通道”的组合。2. 方案选型为什么我最后选了 Flink CDC 这套组合Doris 官方支持很多条同步链路常见的有四种DataX 离线导、Canal 加消息队列加 Routine Load、Flink CDC 直接写 Doris、Doris 外表 Catalog 配合 Insert Into。它们都能用但适用面差别很大。2.1 先说 Doris 自带的几个导入通道Stream Load 是 Doris 最基础的批量导入方式适合把已经准备好的数据文件或内存数据一次性导入。它支持 CSV、JSON支持原子生效失败可以重试。很多离线任务最终都是通过 Stream Load 把数据送进 Doris 的。Routine Load 是 Doris 内置的“订阅消费”能力专为 Kafka 等消息队列准备。它把消费、解析、导入这几个动作做成一个常驻任务只要你把上游 binlog 变化写入 KafkaDoris 就能持续不断消费并写入表里。它不需要额外部署计算引擎部署简单运维成本低。但 Routine Load 有一个短板它只做“搬运”不做复杂转换。如果源端是多库多表合并需要在写入 Kafka 之前就把库表规则、字段补齐、主键拼接这些逻辑处理好。也就是说它适合链路相对简单的“库表一对多”或“多表合并主题”不适合处理灵活多变的库表正则规则。2.2 横向对比几个完整同步架构下面这张表是我做选型时整理的价值在于帮你快速排除不适合自己的方案。同步链路实时性多库多表能力DDL同步开发成本适用场景DataX批量导入 Stream Load分钟到小时级弱需手工配置每个表不支持低离线数仓、T1 场景Canal Kafka Doris Routine Load秒级中需在 Kafka 端做规则弱中简单表同步、主题明确、无大量转换Canal Kafka Flink Doris秒级强Flink SQL 做规则较强中高需要灵活转换、多表合并、复杂过滤Flink CDC Doris Sink秒级强可直接正则匹配多库多表强中高大规模整库同步、分库分表合并Doris External Catalog Insert Into小时级中外表查询需要配置不支持低临时查询、小批量抽数我最终选的是 Flink CDC Doris Sink 这条链路。原因很简单第一Flink CDC 的 source 本身就支持用正则去匹配多个库、多张表天然适合“我不想管它是 01 号库还是 02 号库只要符合规则我就同步”的需求第二Flink 有完整的状态管理和 checkpoint 机制能够做到精确一次语义数据不会丢也不会重复第三Flink CDC 3.x 的整库同步能力支持 DDL 变更同步上游加列、删列能自动作用到下游这一点对表数量多的场景太重要了。2.3 分库分表合并不是简单 UNION ALL做分库分表向 Doris 合并时最容易踩的坑是把问题理解成“上来直接插数据”。实际上你要做的是一套映射规则。假设源端有order_db_01.orders_202501、order_db_01.orders_202502、order_db_02.orders_202501。这些表结构基本一致但可能字段顺序不同、字段名大小写不同甚至某些分表多一个字段。同步到 Doris 时目标表应该是一张逻辑上的大表比如dws_order_info。处理逻辑包括库表匹配用正则order_db_\d和orders_\d匹配所有物理分表字段映射把各分表字段统一映射到目标表字段缺失的字段补默认值主键处理单库单表的主键可能只是业务主键合并时如果不加来源标识会出现两个库同 ID 的数据互相覆盖类型兼容源端 varchar、datetime、decimal 转换为 Doris 对应类型尤其要注意 decimal 精度和字符串长度上限。这套规则放在 Flink SQL 里就是建 source 表、建 sink 表、写一条 Insert Into 的转换逻辑。代码量不大但规则必须提前想清楚否则上线后会出现大量脏数据。3. 实操用 Flink CDC 将上游多库整库同步到 Doris这一部分我按自己的实践流程来讲包含版本、配置、建表、调优照着做基本能把链路跑通。3.1 环境与版本选择我这里使用的版本组合是经过验证的建议不要随便混搭。Doris 2.1 系列表模型用 Unique 主键模型Flink 1.18 系列Flink CDC 3.1 系列连接器选择flink-cdc-pipeline-connector-mysql或 mysql-cdc 相关依赖JDK 8 或 JDK 11注意 Flink 1.18 对 JDK 17 也支持但生产环境我更推荐 JDK 11兼容性更稳如果你是直接用 Flink SQL 开发需要准备这些端到端的依赖flink-connector-mysql-cdc负责从 MySQL binlog 读取增量数据flink-doris-connector负责写入 DorisMySQL JDBC 驱动Doris 客户端的配置fenodes、username、passwordMaven 依赖大致如下版本号以你实际安装的发行版为准dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-mysql-cdc/artifactId version3.1.0/version /dependency dependency groupIdorg.apache.doris/groupId artifactIdflink-doris-connector/artifactId version1.6.0/version /dependency看不懂 Maven 也没关系直接在 Flink 的 lib 目录下放对应 jar 包也可以重点是要保证连接器版本和 Flink 版本匹配否则会出现NoSuchMethodError这类非常恼人的运行时异常。3.2 全量加增量无缝衔接的原理很多第一次用 Flink CDC 的人会问全量同步和增量同步是怎么衔接起来的我简单解释一下。Flink CDC 在读取一张表时会先做一个全量快照这个过程并不是简单地select *。它会先获取一个全局一致的 binlog 位点然后并行读取表数据同时从那个位点开始持续记录新增的 binlog 事件。全量读完之后它会把快照期间产生的增量变更补回来这样就保证了最终从旧数据到新数据的无缝切换。这个过程对多库多表有什么影响影响就是如果你同时同步 200 张表Flink 需要协调好每张表的快照读取和 binlog 位点源库的连接数压力会明显增大。所以配置 source 时一定要控制并行度不要把 source 并行度设成和表数量一样大否则源库瞬间被连接打满DBA 会直接找你。3.3 配置多库多表识别规则Flink SQL 方式下核心是 Source 表的database-name和table-name参数这两个参数都支持正则表达式。比如我要同步所有order_db_开头的库里所有orders_开头的表可以这样定义CREATE TABLE order_source ( id BIGINT, order_no VARCHAR(64), user_id BIGINT, amount DECIMAL(12,2), status INT, create_time TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector mysql-cdc, hostname 192.168.10.10, port 3306, username cdc_user, password your_password, database-name order_db_.*, table-name orders_.*, scan.incremental.snapshot.chunk.size 8096, server-id 5400-5404 );这里有几个细节容易出错。database-name如果填order_db_.*Flink CDC 会把所有匹配到的库中的表都拿过来。但是注意它要求这些表有统一的字段结构因为你定义了同一个 source 表结构。如果各分库字段差异太大建议拆成多个 source 任务而不是硬塞在一起。server-id必须是一个范围并且范围大小要大于等于 source 的并行度。MySQL binlog 同步要求每个并发连接使用不同的 server-id如果你只填一个固定值并行度一提高就会冲突。取值范围不能和线上主从复制的 server-id 冲突这个要和 DBA 确认。目标表 Doris 这边建 Unique 模型表CREATE TABLE IF NOT EXISTS dws_order_info_all ( id BIGINT, order_no VARCHAR(64), user_id BIGINT, amount DECIMAL(12,2), status INT, create_time DATETIME, src_db VARCHAR(64) DEFAULT , src_table VARCHAR(64) DEFAULT ) UNIQUE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 12 PROPERTIES ( replication_num 1 );然后写一条插入语句把源表数据写进目标表INSERT INTO dws_order_info_all SELECT id, order_no, user_id, amount, status, create_time, order_db AS src_db, orders AS src_table FROM order_source;实际项目里你可能需要把src_db和src_table这两个字段加到 source 里用来区分数据来自哪个物理表。3.4 Doris Sink 实际参数调优Flink Doris Connector 的参数非常多我挑几个对稳定性影响最大的说明。CREATE TABLE doris_sink ( id BIGINT, order_no VARCHAR(64), user_id BIGINT, amount DECIMAL(12,2), status INT, create_time DATETIME ) WITH ( connector doris, fenodes 192.168.20.21:8030, username doris_user, password your_password, table.identifier dws.dws_order_info_all, sink.label-prefix doris_sync_order, sink.properties.format json, sink.properties.read_json_by_line true, sink.enable.batch true, sink.max-retries 3, sink.enable-delete true );sink.properties.format和read_json_by_line必须同时配合设置Flink 才会以 JSON 格式逐行发送给 Doris。sink.label-prefix是导入任务的标签前缀Doris 通过 label 做幂等控制。如果 Flink 重启后还用同样的 labelDoris 会直接跳过已导入的数据但注意前缀必须是全局唯一的不同任务不要共用。sink.enable-delete开启后上游 binlog 的 DELETE 事件才能转换成 Doris 的 DELETE 操作。如果你关闭这个参数上游删了一行Doris 里这行还会留着数据就对不上了。开了之后Doris 表必须是 Unique 模型或者 Aggregate 模型且需要function_column.sequence_col做序列列的情况要单独设置。3.5 从“跑通”到“稳定”的配置细节能跑通只是第一步生产环境稳定运行还需要注意几个参数。checkpoint 间隔不能太短。Flink CDC 做的是有状态计算checkpoint 太频繁会导致状态后端频繁快照整个作业的反压会明显上升。我一般设置 10 到 30 秒具体看单条数据大小和吞吐量。source 并行度不要盲目调高。源库 MySQL 的 binlog 只有一个线程在读取时并行度再高也不会线性提升反而增加连接数压力。一般 4 到 8 个并行度足够支撑几十张表。Doris Sink 的sink.flush.queue-size和sink.buffer-size会影响攒批效率。默认值在小数据量下没毛病但如果峰值流量很大建议把 queue-size 设置到 1024 以上buffer-size 设置为 256MB。攒批太大了延迟会高太小了导入小文件过多Doris BE 会频繁做 compaction所以这个值要看实际流量调。最后批式写入不要关。sink.enable.batch为 true 时Flink 会攒一批再发给 Doris吞吐量比逐条写入高很多对 Doris 更友好。4. 同步链路稳定性我踩过的坑和排查方法多库多表同步上线后真正的考验才开始。下面的问题大部分都出现在我实际运维过程中整理成记录帮大家少走弯路。4.1 分库分表主键冲突问题这是多库合并场景里最容易踩的坑。源端每个分库的订单表主键都是自增 ID单库内不重复但合并到 Doris 一张表后order_db_01的 ID100 和order_db_02的 ID100 就会互相覆盖。解决办法有两种。第一种是业务主键本身全局唯一比如订单号是分布式雪花 ID那主键直接用业务主键。第二种是主键本身有重复风险需要在同步时拼接来源标识把组成写入 DSLCREATE TABLE dws_order_info_all ( id BIGINT, src_code VARCHAR(32), order_no VARCHAR(64), ... ) UNIQUE KEY(id, src_code) DISTRIBUTED BY HASH(id) BUCKETS 12;同步时通过函数把库名或分片键放进src_code。这样两条 ID 相同但来源不同的记录在 Doris 里是两条不同记录。如果你的业务需要同一笔订单在 Doris 里只保留一条那就不能简单拼接而要先在 Flink 里做去重再写一个带唯一业务键的目标表。4.2 DDL 同步引发任务失败Flink CDC 3.x 支持同步 DDL比如上游执行ALTER TABLE ADD COLUMN sale_channel INT。DDL 同步确实省事但也会带来很多意外。最常见的问题是上游加了一个非空字段但没有默认值Doris 这边同步目标表却没有自动加列导致写入报错。另一个问题是 Doris 虽然支持部分 Schema Change但某些 DDL 操作比如删除列、修改列类型在同步时可能不支持任务会失败需要手动处理后重启。我的经验是不要完全依赖自动 DDL 同步把它当成辅助能力。更好的做法是提前梳理源端表结构变更规范让上游变更字段时通过工单通知数仓团队由数仓评估后在 Doris 侧先执行对应的 Alter Table再允许上游变更。这个流程看起来很“原始”但对稳定性非常有效。如果确实需要自动同步 DDL我建议在 Flink 侧开启schema.change相关的 pipeline 配置并在任务日志里持续关注DDL executed或schema change failed关键字第一时间收到告警。4.3 上游大事务与延迟告警有一次我负责的同步任务每天固定延迟增长排查半天发现上游每天凌晨跑一次批量更新一次性更新了几百万行binlog 量暴涨Flink 端来不及消费。表现为 source 端堆积、checkpoint 一直完成不了、Doris 导入频率下降。应对大事务有几个思路让上游拆分大事务每批更新不超过 5 万行调大 checkpoint 超时时间比如execution.checkpointing.timeout设为 10 分钟调大 source 端的读取并行度同时调大 Doris Sink 的攒批容量监控指标重点盯两个Flink 的busyTimePerSecond或反压指标以及 Doris 的SHOW LOAD里导入任务的平均落库耗时。一般发现延迟增加先看是 source 端还是 transform 端还是 Doris Sink 端。最简单的方法是看 Flink UI 里每个算子的 BackPressure 和 Records Sent往前端定位不要一上来就盲目加资源。4.4 常见问题速查表下面这张表是我自己排查时常用的备忘录直接照查。现象可能原因排查思路同步任务启动后无数据源库表正则没匹配上或账号无权限先用命令行测试正则匹配确认匹配到了哪些表数据有重复Doris 表用了 Duplicate 模型或未开启 delete 同步检查目标表模型改成 Unique 模型开启 delete 支持更新不生效主键字段设置错误或没有来源标识字段检查主键是否全局唯一必要时加联合主键DELETE 不生效sink.enable-delete没开启开启该参数确认模型为 Unique 模型任务频繁重启server-id 冲突或权限配置错误检查 server-id 范围、MySQL binlog 权限上游加列后同步失败DDL 同步未开启或 Doris 不支持该 DDL手动 Alter Doris 表重启任务从位点恢复Doris 导入报 Fe 连接失败Fenodes 配置不对或账号权限不足检查链接地址、端口、用户权限延迟持续增长大事务、checkpoint 过慢、sink 批量太小看 Flink UI 反压、checkpoint 时长、源库压力4.5 上线前必检查清单以下清单根据实战经验整理强烈建议打印出来逐条打勾。源库 binlog 已经开启且保留时间大于 24 小时CDC 账号有SELECT、RELOAD、SHOW DATABASES、REPLICATION SLAVE、REPLICATION CLIENT权限server-id 范围和现有复制链路不冲突Doris 表已建好Unique 模型、分桶键、副本数合理目标表字段类型和源表字段类型做过兼容性映射sink.enable-delete已按需开启所有必要的监控指标已配置包括 Flink 反压、checkpoint 失败数、Doris 导入失败率回滚方案已准备包括停掉同步任务、保留 binlog 位点、可重新从某个时间点补数据。5. 如果不想引入 Flink还有什么轻量替代方案不是所有团队都有能力维护 Flink 集群。如果你的场景没那么复杂可以用下面两种轻量方案。5.1 基于 Doris Routine Load 的消息队列方案最简单的结构是源库 MySQL 通过 Canal 监听 binlog把变更记录以 JSON 格式写入 Kafka 的某个 topicDoris 使用 Routine Load 直接消费这个 topic。整个过程只需要部署 Canal 和 KafkaDoris 自己就能完成消费导入。创建 Routine Load 的语句类似这样CREATE ROUTINE LOAD dws_order_load ON dws_order_info_all COLUMNS(id, order_no, user_id, amount, status, create_time) PROPERTIES ( desired_concurrent_number 3, max_batch_interval 20, format json ) FROM KAFKA ( kafka_broker_list 192.168.30.11:9092, kafka_topic ods_order_info, kafka_group_id doris_order_group );这个方案的优点是部署简单不引入 Flink运维压力小。缺点是如果源端是多库多表你需要在 Canal 到 Kafka 之间做一层“统一格式转换”把不同库、不同表的记录都转成目标表期望的 JSON。新增一张分表时还要确保 Canal 订阅的规则覆盖到它。DDL 变更同步基本很难自动处理一般要人工介入。它适合表数量比较少、规则稳定的同步链路。5.2 基于 Canal 加自研消费程序有些团队会用 Java 或 Go 写一个后台服务从 Kafka 拉取 Canal 的 binlog 消息解析之后调用 Doris Stream Load 接口写入。这种方式灵活性很高你可以完全控制字段拼接、路由、异常处理。但是从我的经验看这种自研消费程序坑很多消费位点怎么保存、服务重启怎么恢复、消息重复怎么幂等、Doris 写入失败怎么重试、上游 DDL 怎么感知。每一样都需要投入时间去打磨如果没有专门的实时团队我建议不要轻易走这条路。除非你就是要做一个非常定制化的同步比如同步过程中要调用外部接口做数据加工否则完全可以用 Flink 代替。5.3 到底怎么选我的判断标准很简单如果你只有几张表且表结构很稳定用 Canal 加 Kafka 加 Routine Load轻量稳定出了故障也容易排查。如果你有几十张甚至几百张表库表会动态扩展还涉及分库分表合并、字段补齐、复杂过滤那就不要犹豫直接用 Flink CDC。前期学习成本高一点但后续新增一张表、加一个库都只是改个正则的事。如果你只是每天跑一次离线任务延迟要求不高用 DataX 或者 Doris 的INSERT INTO ... SELECT就够了没必要为它搭一套实时链路。6. 最后分享一点个人经验做了这么多同步链路我最大的体会是多库多表同步的难点从来不是工具不会用而是你对源端数据资产的理解不够。每个库的表结构差异、字段语义、主键分布、数据量级都要在开工之前梳理清楚。我曾遇到一个项目开发同学花了两周搭 Flink 链路结果上线前一天才发现两个分库的user_id字段语义不一样一个是用户 ID一个是商户 ID所有关联统计全部要返工。我建议你拿到这个需求后先不要急着写代码先做三件事。第一把源端所有库表列出来标记同名表、异构表、分表第二确认主键是否全局唯一不唯一就设计来源标识第三先挑两张核心表跑一个全量加增量的试点验证整条链路稳定后再扩容到所有表。另外有一个小技巧同步任务上线初期日志和监控一定要看得仔细。Flink CDC 的启动日志会打印它匹配到了哪些库表先确认这个列表和预期一致。Doris 那边用SHOW LOAD能看到每个导入任务的状态异常时 Doris 会返回具体错误行内容不要只看失败计数把错误内容拉出来看是类型转换问题还是列数不匹配。最后生产环境一定要预留回滚方案。我的习惯是同步任务运行前记录好 binlog 位点一旦出现大规模数据异常可以停掉任务清空 Doris 对应表分区再从最近一个位点重新消费。这个操作听起来麻烦但在关键时刻能救命。