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 要看的是分钟级甚至秒级的实时数据，离线 T+1 或者十分钟一次根本无法覆盖。

Doris 本身不是一个数据抓取工具，它是一个高性能的分析型数据库。它提供了 Stream Load、Broker Load、Routine Load 这些导入通道，但是“从源端数据库里把变化数据抓出来”这件事，必须搭配一套 CDC（Change 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 变化写入 Kafka，Doris 就能持续不断消费并写入表里。它不需要额外部署计算引擎，部署简单，运维成本低。

但 Routine Load 有一个短板：它只做“搬运”，不做复杂转换。如果源端是多库多表合并，需要在写入 Kafka 之前就把库表规则、字段补齐、主键拼接这些逻辑处理好。也就是说，它适合链路相对简单的“库表一对多”或“多表合并主题”，不适合处理灵活多变的库表正则规则。

2.2 横向对比几个完整同步架构

下面这张表是我做选型时整理的，价值在于帮你快速排除不适合自己的方案。

我最终选的是 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：负责写入 Doris
• MySQL JDBC 驱动
• Doris 客户端的配置（fenodes、username、password）

Maven 依赖大致如下，版本号以你实际安装的发行版为准：

<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-mysql-cdc</artifactId> <version>3.1.0</version> </dependency> <dependency> <groupId>org.apache.doris</groupId> <artifactId>flink-doris-connector</artifactId> <version>1.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 重启后还用同样的 label，Doris 会直接跳过已导入的数据；但注意前缀必须是全局唯一的，不同任务不要共用。

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的 ID=100 和order_db_02的 ID=100 就会互相覆盖。

解决办法有两种。

第一种是业务主键本身全局唯一，比如订单号是分布式雪花 ID，那主键直接用业务主键。第二种是主键本身有重复风险，需要在同步时拼接来源标识，把组成写入 DSL：

CREATE 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 常见问题速查表

下面这张表是我自己排查时常用的备忘录，直接照查。

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 的某个 topic，Doris 使用 Routine Load 直接消费这个 topic。整个过程只需要部署 Canal 和 Kafka，Doris 自己就能完成消费导入。

创建 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 对应表分区，再从最近一个位点重新消费。这个操作听起来麻烦，但在关键时刻能救命。