上周一个搞实时数仓的朋友跑来找我说他们用 Doris 已经两个月单表同步跑得很顺可一旦要把线上二十多个业务库、八九百张表一次性接进来整个人就麻了。这个场景我太熟了——“Doris 同步多库多表”这几个字几乎每个想把 Doris 真正用起来的数据团队都会撞上。业务系统按微服务拆库、按时间或 ID 分表实时报表和跨库关联分析却要求把所有需要的表持续汇进 Doris单表同步那套经验根本撑不住。这篇文章我就把这些年的完整实践写下来从选型怎么定、Flink CDC 和 Doris Sink 怎么配到新增表如何不停机接入、性能怎么调、哪些坑反复踩最后再聊聊什么场景其实不该用 CDC。内容偏实操适合正在做 Doris 实时数仓、或者准备从单表同步走向整库同步的工程师看完可以直接照着搭一套。1. 为什么“多库多表”同步是 Doris 上生产的第一道坎1.1 业务侧的真相没有“几张表”是几十个库、上千张表很多团队一开始理解错了需求。Doris 上线初期往往挑核心的订单表、用户表做同步几个任务跑起来POC 演示很顺利。可真到生产接入阶段你会发现业务侧的存量远超想象一个电商项目订单库、用户库、商品库、营销库、库存库各是各的库订单表还要按月分表、按用户 ID 取模分表一个游戏项目区服分库、日志分表几十个 MySQL 实例稀松平常。你真正要面对的不是“同步几张表”而是“把几百上千张表持续、可靠、低延迟地汇入 Doris”。为什么要汇因为 Doris 的价值就是统一查询。业务方想算“用户最近 30 天在各渠道的下单总额”这个查询横跨订单、用户、渠道三张表而且每张表都来自不同的上游库。如果这些表不集中到 Doris分析就只能回源 MySQL 做跨库 Join性能和稳定性都扛不住。所以多库多表同步不是可选项是实时数仓的入场券。1.2 三种接入姿势成本差异是数量级的接触过不少团队接入方式无非三种我把它们的差异整理成了一张表接入方式典型工具延迟适合场景真正的成本定时批量同步DataX、SeaTunnel 批模式T1 或小时级初始化全量、离线数仓分层无增量能力每次跑全量慢单表单任务 CDC每张表一个 Flink 任务秒级几张核心表上千张表 上千个任务运维爆炸多库多表统一 CDCFlink CDC 多表/YAML Pipeline秒级整库整表持续入仓依赖配置规范、DDL 策略、监控体系单表单任务的问题做过的人都懂每张表要维护建表语句、任务启动命令、重启恢复位点上游表结构一变更几百个任务逐个排查。最初跑十几张表没感觉一旦上百张每周光处理任务故障就耗掉大半天。多库多表统一拉通看着只是换了个写法实际上是把“运维单表”变成了“运维一套规则”这是数量级的差距。1.3 动手前必须先回答的三个问题第一源是什么。同步方案几乎都是围绕 MySQL 生态展开的如果你的上游是 PostgreSQL、Oracle、SQLServerFlink CDC 也有对应 connector但配置细节、支持程度不一样别指望一套配置全搞定。第二目标表用哪种 Doris 模型。实时同步上游业务表几乎无脑选 Unique 模型用主键去重保证幂等只有明确要做历史明细留存的场景才用 Duplicate 模型。Aggregate 模型不建议放在同步层预聚合结果一旦口径变化很难纠正。第三要不要同步结构变更。这是最容易被忽略的点。很多团队上线时把表结构对好之后上游加了个字段同步任务要么写不进、要么悄悄丢数据三个月后对账发现对不上再回头补就痛苦了。这个问题没有银弹但必须有预案后面第三章专门讲。2. 主流的实现路径Flink CDC Doris Sink 的组合打法2.1 组件版本怎么配先把这个搞对多库多表同步最稳的组合目前就是 Flink CDC 做采集、Flink Doris Connector 做写入。但版本匹配是第一个大坑我见过太多人栽在版本错配上用了老版本的 Doris Connector 配新 Flink作业提交直接 ClassNotFound或者 Flink CDC 版本太老不支持正则匹配多表整库同步根本做不了。一个我验证过、跑得比较省心的组合组件推荐版本说明Apache Flink1.17.x / 1.18.x长期维护版本生态兼容好Flink CDC2.4 或 3.x2.x 走 Flink SQL3.x 支持 YAML Pipeline 整库同步flink-connector-doris1.5.0匹配 Flink 1.17支持两阶段提交Apache Doris2.0.x / 2.1.xUnique 模型和 Stream Load 2PC 能力强版本选择上我的原则是“在主线版本里选新的”。Flink CDC 3.x 的多表同步能力比 2.x 强很多如果是从零开始直接上 3.x如果团队已经有一堆 2.x 的存量任务不必立即迁移但新任务建议直接用 3.x 的 Pipeline 方式。2.2 方式一Flink SQL用正则一次匹配多张表Flink SQL 方式的核心是利用 MySQL CDC Connector 的database-name和table-name支持正则匹配。举个例子我要把trade_db_0到trade_db_3四个库里的order_开头的所有分表同步进来源表可以这样定义CREATE TABLE cdc_orders ( id BIGINT, 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.20, port 3306, username cdc_user, password your_password, database-name trade_db_[0-9], table-name order_[0-9], server-id 5401-5420, scan.incremental.snapshot.chunk.size 8192, scan.startup.mode initial );这里两个点要解释清楚。database-name和table-name是正则而非普通通配符trade_db_[0-9]能同时命中trade_db_0到trade_db_9order_[0-9]能命中所有按月或按 ID 后缀的分表。server-id必须给一个范围而不是单值因为多并行度下每个并发都需要一个独立 server-id给单值会导致读取源库 binlog 时被其他同步任务挤下线这是个高频坑。目标表同样建在 DorisCREATE TABLE doris_orders ( id BIGINT, user_id BIGINT, amount DECIMAL(12, 2), status INT, create_time TIMESTAMP(3) ) WITH ( connector doris, fenodes 192.168.10.30:8030, table.identifier ods.trade_orders, username doris_user, password your_password, sink.label-prefix ods_trade_orders, sink.properties.format json, sink.properties.read_json_by_line true, sink.enable-2pc true, sink.max-retries 3 );然后一句INSERT INTO doris_orders SELECT ... FROM cdc_orders就把多张分表的数据汇到 Doris 的一张 Unique 表里。这个方式的限制是正则匹配的多张表结构必须一致且目标只能是一张表。如果每张表结构各不相同、目标也要一一对应用 Flink SQL 就得每张表写一对源和目的的 INSERT本质上又退回“单表模式”了。真要解决“多库多表、异构入库”得看下一种方式。2.3 方式二Flink CDC 3.x YAML Pipeline整库整表一把梭Flink CDC 3.x 引入的 Pipeline 模式是专门解决多库多表同步的。它用一个 YAML 描述“从哪来、到哪去、怎么跑”一次提交就能同步一批库表还能自动在 Doris 建库建表。看一个实际可用的配置source: type: mysql hostname: 192.168.10.20 port: 3306 username: cdc_user password: your_password tables: trade_db_.*\.order_.* server-id: 5401-5430 chunk-size: 1024 sink: type: doris fenodes: 192.168.10.30:8030 username: doris_user password: your_password properties: format: json read_json_by_line: true schema.change: true pipeline: parallelism: 2tables这里用正则把几十个库、上百张分表一口气圈进来Flink CDC 会自动为每张匹配到的表建 Doris 目标表并持续同步。schema.change: true打开后上游加列这类基础变更能尝试自动同步到 Doris。这个模式最大的价值就是“自动发现”后面新增一张表只要名字符合正则任务无需重启就能把它纳入同步范围。对几百张表的规模来说运维成本比 Flink SQL 那种逐表 INSERT 低一个量级。2.4 从提交到验证跑通任务的操作清单配置写好只是第一步我把从提交到验证的完整动作列出来照着做基本不会漏把flink-connector-doris和flink-cdc相关 jar 放进 Flink 的lib目录注意版本对应。用flink run提交 Pipeline 作业建议指定一个固定 Job 名称方便后续定位。到 Flink Web UI 看作业是否进入 Running重点看 Source 是否已经读到记录数。在 Doris 里执行SHOW TABLES确认表已自动创建SELECT COUNT(*)对比源库行数。在源库手动做一条 INSERT、一条 UPDATE、一条 DELETE等几秒后在 Doris 侧验证数据是否变化。观察 checkpoint 是否稳定完成检查间隔建议默认先跑 10 秒再根据压力调整。一套走下来没问题才算真正跑通可以进入下一步的“生产化加工”。3. 生产环境最关键的“自动发现”新增表、改表结构怎么不停机接入3.1 新增库、新增表的接入方式多库多表同步上线后第一个现实问题就是业务方这周又建了一张新表怎么接进来最省事如果你的同步任务从一开始就是用正则匹配的比如trade_db_.*\.order_.*那么新表只要命名符合规则Flink CDC 会自动发现并把它纳入同步不需要任何人工操作。前提是开启并确认了 Pipeline 的自动建表能力。这是我在生产上最推荐的做法它把“新增表”这件事从“发工单、改配置、重启任务”变成“业务方自己按规范建表就行了”。如果新表命名不满足现有正则那就要改任务配置。正确做法不是直接强杀任务重启而是先触发一次 savepoint然后修改配置、用-s savepoint路径从状态恢复启动。这样全量阶段不会重跑增量数据也不会丢。命令大概是# 触发 savepoint flink stop job_id -p /data/flink/savepoints # 修改配置后从 savepoint 恢复 flink run -s /data/flink/savepoints/savepoint-xxx -c 你的入口类 your-job.jar这一步值得多说一句很多人图省事直接flink cancel再启动如果源表是全量增量混合读取恢复时会重新走全量扫描几千张大表能把源库压垮运维也因此背过锅。savepoint 恢复虽然慢一点但是稳。3.2 DDL 变更上游加了个字段Doris 怎么办这是多表同步里最头疼的环节。MySQL 的业务表加字段太常见了“加个备注”“加个标记位”一天能出好几个变更。Doris 侧的同步逻辑对 DDL 的处理远没有大家想象的“自动”Flink SQL 模式下如果源表新增字段而目标表没有作业往往会报 schema 不匹配或写入异常如果目标表先加了字段而源表还没加Doris 写入时多出的字段又可能因为列映射对不上而报错。Flink CDC 3.x Pipeline 模式下schema.change对“加列”这类变更支持相对好但“修改字段类型”“删列”这类操作基本还是要人工介入。我的建议是分层处理加列尽量由同步任务自动同步或者执行一个原子的 ALTER TABLE 让任务短暂重建。改类型比如 varchar 变长、int 变 bigintDoris 侧需要评估兼容性一般要停写、ALTER、再恢复不能指望 CDC 全自动。删列默认不同步让 Doris 保留冗余字段等大版本维护窗口统一处理。这里最忌讳的做法是“源库 DDL 和生产任务完全脱节”。哪怕你有 DBA 审批流程也一定要把 Doris 同步任务纳入变更通知范围否则三个月后对账发现数据和结构都对不上回查成本高到你想哭。3.3 自动建表的辅助手段元数据生成 DDL如果你的同步方案没有自动建表能力比如还在用 Flink SQL 每表手动建那批量建表可以靠元数据生成。从 MySQL 的information_schema把表结构拉出来生成 Doris 的建表语句再用脚本批量执行能省掉大量重复劳动。核心思路是这样一段查询SELECT CONCAT( CREATE TABLE IF NOT EXISTS ods_db., TABLE_NAME, (, GROUP_CONCAT( CONCAT(COLUMN_NAME, , CASE WHEN DATA_TYPE bigint THEN BIGINT WHEN DATA_TYPE int THEN INT WHEN DATA_TYPE IN (varchar, char) THEN VARCHAR(||CHARACTER_MAXIMUM_LENGTH||) WHEN DATA_TYPE decimal THEN DECIMAL(||NUMERIC_PRECISION||,||NUMERIC_SCALE||) WHEN DATA_TYPE datetime THEN DATETIME ELSE STRING END) ORDER BY ORDINAL_POSITION SEPARATOR , ), ) UNIQUE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 10 PROPERTIES(replication_num 1); ) FROM information_schema.columns WHERE table_schema trade_db_0 GROUP BY table_name;生成的语句大多可以直接在 Doris 执行遇到复杂的列定义再手工调整。注意这只是建表数据还是得靠同步任务灌入别把建表和同步混为一谈。4. 性能调优与一致性从“能跑”到“跑得稳”4.1 这几个参数决定同步不白干任务能跑通和任务跑得稳、跑得快中间隔着参数调优。多库多表同步涉及的参数很多挑影响最大的几个说。参数建议值作用与注意点pipeline.parallelism2 起步按表数量加到 8并行度高吞吐大但会消耗更多源库连接和 server-idcheckpoint 间隔默认 10s延迟敏感设 5s直接决定 Doris 端可见延迟因为两阶段提交在 checkpoint 时触发scan.incremental.snapshot.chunk.size4096~8192全量阶段分片大小太大单线程压力大太小任务列表膨胀sink.max-retries3Stream Load 失败重试次数别设太大否则异常积压sink.enable-2pctrue开启 Doris Stream Load 两阶段提交保证精确一次有一个内在关联很多人没意识到checkpoint 间隔越短Doris 写入越频繁、延迟越低但 Flink 做 checkpoint 的开销也越大。如果你的源库表非常多checkpoint 频繁做会导致状态后端压力大建议从 10 秒开始观测一段时间再调低不要一上来就追求极致延迟。4.2 全量阶段和增量阶段优化重心完全不同Flink CDC 读取 MySQL 时initial模式会先做全量快照再无缝切到增量 binlog。这两个阶段的表现特征完全不一样调优也不能一套方法用到底。全量阶段瓶颈通常在源库。快照扫描会给 MySQL 带来额外压力并发太高会把业务库拖慢。这时候应该控制扫描并发适当调大chunk-size减少任务分片数量同时告诉业务方“全量初始化期间源库压力会上升”最好安排在低峰期。增量阶段瓶颈通常在你自己的消费链路。Flink Web UI 上如果看到 Source 出现背压但 Doris Sink 没有明显瓶颈往往是server-id不够用或者并行度不匹配。如果 Sink 出现背压多半是 Doris 侧导入频率跟不上要么调大并行度要么适当放宽sink.properties里的 flush 间隔。判断当前处于哪个阶段很简单看 Flink Web UI 上 Source 的“records read”增长斜率全量阶段是阶梯式猛增增量阶段是平稳小浪涌。4.3 端到端一致性两阶段提交和 Unique 模型是怎么配合的多库多表同步最怕“数据坏了还不知道”。FlinkDoris 的组合之所以稳靠的是两层配合Flink connector 在 checkpoint 时向 Doris 发起 Stream Load 两阶段提交先预提交、checkpoint 成功后正式提交保证“要么都成功要么都不成功”同时 Doris 目标表用 Unique 模型主键相同的数据后到覆盖先到天然幂等。两套机制叠在一起就实现了精确一次语义。但这里有个前提目标表必须有明确主键。如果你的源表没有主键或者主键选择不当Unique 模型的去重效果就打了折扣重复数据可能悄悄累积。这一点在生产上非常要命后面踩坑部分细说。另一个关键点是“删除怎么同步”。MySQL 里物理 DELETE 的行binlog 里是删除事件Doris 的 Stream Load 默认只做导入不会自动把对应主键删掉。所以要么在同步链路里显式处理删除标记要么更推荐把上游的物理删除改造成逻辑删除加一个is_deleted状态字段这样 Doris 侧只需更新状态不用真删数据。这个改造对实时数仓利大于弊“只增不改删”能让所有下游分析口径都保持可追溯。5. 踩坑实录多表同步里反复出现的几个大坑5.1 源库 Binlog 配置不当任务直接连不上多库多表同步的源头是 MySQL binlog但很多 MySQL 实例的默认配置根本不是为 CDC 准备的。我遇到过的典型问题是任务一启动就报“Could not find first log file name in binary log index”或者读到一半连接断掉。排查下来大多是这几项没满足binlog 格式必须为 ROWbinlog_format ROW。binlog_row_image FULL否则更新事件里只有变更列默认值和其他列对不上。同步账号要有SELECT、REPLICATION SLAVE、REPLICATION CLIENT权限命令如下GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO cdc_user%;如果你所在公司的 MySQL 实例很多是 DBA 统一托管的建议提前把“开通 CDC 同步的最小配置清单”发给 DBA一次说清别让业务方反复催。还要确认 binlog 保留时长默认可能只有 1 天一旦同步任务宕机超过保留窗口重启后无法从旧位点继续只能回退全量重导这是最惨的恢复场景。5.2 server-id 冲突任务跑一会儿就断连这个坑我几乎每次都要提醒。现象很诡异任务刚启动好好的跑几分钟到几小时不等Source 突然报连接被重置看 MySQL 的错误日志发现同一个 server-id 被多个连接同时使用。原因很简单——多库多表同步有多个并行子任务每个并行度都需要一个独立的 server-id而不同团队的不同任务如果都用了默认的server-id就会互相“挤占”。解法也很直接给每个同步任务分配一个足够大的、不重叠的 server-id 范围。我的习惯是从 5400 开始按任务递增每个任务给 20~30 个 id。比如任务 A 用5401-5420任务 B 用5421-5440。这个规范要写进团队的接入文档新任务照着填就不会出现“玄学断连”。5.3 源表没有主键Doris 主键从哪来前面说过 Unique 模型必须有主键但业务库里的表并不总是有主键。日志表、流水表、中间表经常就是纯 append 风格MySQL 层没有定义任何主键。这时候你不能直接让同步任务“没有主键也硬建 Unique 表”否则 Doris 端去重会变成全列去重性能差、语义还不对。实操上有两种做法。第一种用业务自然键拼接主键比如“用户ID 时间 唯一流水号”拼成一个字段。第二种如果确实找不出稳定主键就把目标表建成 Duplicate 模型保留明细需要去重时再在 Doris 里用视图或物化模型做。我的建议是先和业务确认表的真实语义别急着拍板用 Duplicate因为一旦上了线后续想从 Duplicate 迁到 Unique 要重导全量数据代价很大。5.4 时区、精度与类型映射的隐性差异多表同步的数据质量往往不是坏在主链路而是坏在这些“小字段”上。MySQL 的datetime没有时区概念但 CDC 读取时会按连接串里的serverTimezone解释如果你的 Doris 任务和数据平台默认时区不一致所有时间字段都会偏移 8 小时或更多。同步前后一定要拉一个时间字段做抽样对比。精度问题更容易阴沟翻船。MySQL 的decimal(20,4)在 Doris 里要用高精度 DECIMALV3 才能完整表达varchar(2000)到 Doris 侧如果建成了VARCHAR(255)写入超长会直接失败或截断。所以自动建表时文本字段长度要按源库真实最大值加冗余宁可大一点也别让同步链路因为一个超长字段整个任务报错。5.5 任务重启与恢复savepoint 用不好等于白干最后再说重启。生产上难免要升级版本、改配置、调并行度任务重启是常态。但重启方式决定了你的数据会不会重、会不会丢。用 savepoint 恢复有两个隐雷。一是 Doris Sink 的sink.label-prefix如果和之前任务完全一样恢复后 Stream Load 可能因为 label 重复被 Doris 拒绝报“Duplicate label”。解法是给 label 前缀加上版本号或时间戳每次变更配置时顺手换个前缀。二是 checkpoint 状态过大时恢复时间很长如果目标只是“尽快恢复增量”可以接受少量数据回溯那用scan.startup.mode latest-offset直接从当前 binlog 位点继续牺牲重建状态的时间换来快速上线。业务容忍度不同选型也不同但一定要知道这两个选项各自意味着什么。6. 什么时候别用 CDC跨集群复制与离线兜底6.1 Doris 到 Doris用 Doris 自己的跨集群能力不是所有多库多表同步都是从 MySQL 到 Doris。如果源和目标都是 Doris——比如要做主集群和备集群的数据一致、做集群迁移、或者把在线 Doris 的数据复制到另一个分析集群——这时候用 Flink CDC 就不合适了属于拿牛刀杀鸡还杀不好。Doris 自身提供了跨集群复制和备份恢复的能力专门处理 Doris 实例间的数据同步。判断标准很简单源是业务 MySQL用 Flink CDC源是 Doris用 Doris 自身的复制机制。把这两种场景分清能避免很多不必要的复杂度。跨集群复制的优势是它直接在 Doris 内部走元数据和数据副本的同步表结构变更也能一并处理不需要你维护一套 Flink 集群。6.2 大批量历史数据初始化先离线灌全量再起 CDC多库多表同步最容易出的生产事故就是“全量阶段把源库搞挂了”。几百张表同时走 initial 模式全量扫描哪怕限制了并行度对源库的瞬时压力也很大。我的习惯是走两段式先用离线方式把历史数据灌进 Doris再启动 CDC 从 binlog 位点追增量。具体做法是在启动 CDC 任务之前先在源库执行SHOW MASTER STATUS记下当前的 binlog 位点用 DataX 或 Stream Load 把全量数据导入 Doris然后启动 CDC 任务时指定scan.startup.mode为指定位点GTID 或文件名位置。这样全量压力集中在离线导入工具上可控增量追平也很快。前提是 binlog 保留时间必须覆盖“全量导入”这个窗口否则追增量时会发现位点已过期前面的工作白做。6.3 规模不大时别一上来就堆三套集群最后泼盆冷水如果你的表总量就几十张数据量一天几个 GB不需要为了“多库多表”专门搭一套高可用的 Flink 集群。单机 Flink Standalone 加上 Flink CDC Pipeline 模式完全够用省下的机器成本可以干别的。几百张表再考虑独立集群、独立状态后端。资源给多了维护复杂度也上去这个平衡要靠实际规模和增长趋势来判断不是配置越豪华越好。多库多表同步这件事我踩过最深的坑其实是“把同步当成一次性建表任务”。真正的难点不在第一天的全量接入而在之后每一天的新表、新字段、新故障。把正则匹配、自动建表、savepoint 恢复、DDL 预警这套机制建好几百张表的同步也就是一套配置的事。最后再分享一个小技巧给每个同步任务起一个规范的 Job 名称比如cdc_mysql_to_doris_trade_2024label 前缀跟着任务名走出问题查日志的时候能少花一半时间。