MySQL CDC 直接写进 Doris 汇总模型同步即汇总明细不用搬两遍很多实时链路是这样做的先把 MySQL 明细一条条同步到 Doris再起一个定时任务做GROUP BY把结果写进汇总表。链路长、延迟高、还多一份明细存储。这次我们换一种做法CDC 进来什么Doris 在写入时就直接汇总掉。一、为什么要把“汇总”放到写入时做先看一个很常见的实时大屏需求按「日期 商品」看订单件数和销售金额数据要准到秒不能等 T1明细表每天几百万行但看板只关心几十个维度组合最好别再维护一套定时聚合任务。传统做法分两步MySQL 明细 ──CDC── Doris 明细表 ──定时 GROUP BY── Doris 汇总表 ── 看板问题也很明显多存一份明细Doris 里既有明细又有汇总存储翻倍多一段延迟汇总任务没跑完看板就是旧数据多一套任务调度、补偿、口径维护都是成本明细越大越慢每次聚合都要扫一遍新增明细。而 Doris 的聚合模型Aggregate Key天生就是为这件事准备的MySQL 明细 ──CDC── Doris 聚合模型 ↑ 写入时自动按 Key 合并同一个 Key 的多行数据在写入阶段就被SUM/REPLACE掉。查询时看到的已经是汇总结果不需要再扫明细。这篇文章做一次完整验证源端MySQL 8.0开启 ROW 模式 Binlog链路DataY 的MySQLBinlogInput→DorisStreamLoad目标端Doris 2.1AGGREGATE KEY汇总模型验证内容全量快照、实时增量、批量写入三条路径是否都能自动汇总。二、先看结果1010 行明细Doris 里只剩 17 行先把最终结果放出来。源表 MySQLorder_detailsource_rows agg_keys source_qty source_amount 1010 17 1018 105432.00Doris 聚合表order_aggagg_rows total_qty total_amount 17 1018 105432.001010 行明细按「日期 商品」聚合后只剩 17 行件数和金额与源端完全一致。其中 10 个批量商品每个商品写了 100 行Doris 里每个商品只剩 1 行stat_date product_id product_name order_cnt sales_amount 2026-10-27 2001 批量商品2001 100 9990.00 2026-10-27 2002 批量商品2002 100 9990.00 ... 2026-10-27 2010 批量商品2010 100 9990.00这就是聚合模型的价值汇总不是查出来的是写进去的。三、Doris 聚合模型Key 相同就自动合并先建目标表CREATEDATABASEIFNOTEXISTScdc_demo;CREATETABLEcdc_demo.order_agg(stat_dateDATENOTNULLCOMMENT统计日期,product_idINTNOTNULLCOMMENT商品ID,product_nameVARCHAR(100)REPLACECOMMENT商品名称,categoryVARCHAR(50)REPLACECOMMENT品类,order_cntBIGINTSUMDEFAULT0COMMENT订单件数,sales_amountDECIMAL(18,2)SUMDEFAULT0COMMENT销售金额,last_updateDATETIMEREPLACECOMMENT最后更新时间)AGGREGATEKEY(stat_date,product_id)DISTRIBUTEDBYHASH(product_id)BUCKETS3PROPERTIES(replication_num1);三个关键点AGGREGATE KEY(stat_date, product_id)这两列是聚合维度。相同「日期 商品」的行会被合并成一行。SUM指标列用SUM。每来一批数据就在已有值上累加。REPLACE维度列用REPLACE。同一个 Key 再来一行时用最新值覆盖旧值。所以last_update会自动保留最后一次变更时间。如果用DESC看表结构会看到每个非 Key 列都带着聚合方式Field Type Key Extra stat_date DATE true NULL product_id INT true NULL product_name VARCHAR(100) false REPLACE category VARCHAR(50) false REPLACE order_cnt BIGINT false SUM sales_amount DECIMAL(18,2) false SUM last_update DATETIME false REPLACE注意聚合模型要求先建表。DataY 的DorisStreamLoad支持目标表不存在时自动建表但自动建出来的是按源表结构映射的普通模型不会自动变成AGGREGATE KEY。想做写入时汇总目标表要提前按业务口径建好。四、源表与测试数据MySQL 源表保持“订单流水”的形态一行代表一次销售CREATEDATABASEcdc_demoDEFAULTCHARACTERSETutf8mb4;CREATETABLEcdc_demo.order_detail(idBIGINTPRIMARYKEYAUTO_INCREMENT,stat_dateDATENOTNULL,product_idINTNOTNULL,product_nameVARCHAR(100),categoryVARCHAR(50),order_cntINT,sales_amountDECIMAL(10,2),last_updateDATETIME)ENGINEInnoDBDEFAULTCHARSETutf8mb4;写入 7 条初始数据INSERTINTOorder_detail(stat_date,product_id,product_name,category,order_cnt,sales_amount,last_update)VALUES(2026-10-25,1001,智能音箱 Pro,数码,1,399.00,2026-10-25 10:00:00),(2026-10-25,1002,机械键盘,数码,2,598.00,2026-10-25 10:05:00),(2026-10-25,1001,智能音箱 Pro,数码,1,399.00,2026-10-25 11:00:00),(2026-10-25,1003,手冲咖啡壶,家居,3,267.00,2026-10-25 12:00:00),(2026-10-26,1001,智能音箱 Pro,数码,2,798.00,2026-10-26 09:00:00),(2026-10-26,1002,机械键盘,数码,1,299.00,2026-10-26 09:30:00),(2026-10-26,1003,手冲咖啡壶,家居,2,178.00,2026-10-26 10:00:00);这 7 行里2026-10-25 1001出现了两次。源端是两行Doris 里应该只剩一行件数变成 2金额变成 798。这就是第一个验证点。五、DataY 任务一个 CDC 输入一个 Doris 输出任务配置只用了两个组件{units:[{.id:mysql_cdc,.name:MySQLBinlogInput,datasource:{type:mysql,url:jdbc:mysql://127.0.0.1:3306/cdc_demo?useSSLfalseallowPublicKeyRetrievaltrueserverTimezoneAsia/Shanghai,driver:com.mysql.cj.jdbc.Driver,username:root,password:******,dbschema:cdc_demo},databaseNamePattern:cdc_demo,tableNamePattern:order_detail,serverId:1234,snapshot:true},{.id:doris_sink,.name:DorisStreamLoad,datasource:{type:Doris,url:jdbc:mysql://127.0.0.1:9030/cdc_demo?useSSLfalseallowPublicKeyRetrievaltrueserverTimezoneAsia/Shanghai,driver:com.mysql.cj.jdbc.Driver,username:root,password:******,dbschema:cdc_demo,extraParams:{fe_endpoint:http://127.0.0.1:8030}},schema:cdc_demo,table:order_agg,model:insert}],connections:[{sourceId:mysql_cdc,targetId:doris_sink}]}几个参数值得单独说参数作用snapshot: true第一次运行时先读历史全量再从当前 Binlog 位点接增量databaseNamePattern只采集cdc_demo库tableNamePattern只采集order_detail表serverId伪装成从库的 ID同一 MySQL 实例多个任务要互不相同fe_endpointDoris FE 的 HTTP 地址Stream Load 走这个入口model: insert以 INSERT 语义写入聚合模型会自动完成合并启动方式java-jardatay-core-1.0.4-jar-with-dependencies.jar cdc-task.json日志里会看到两个阶段开始读取历史全量数据snapshot起始位点 mysql-bin.000003:6464 历史全量数据读取完成表 cdc_demo.order_detail共 7 条 历史全量数据读取完成启动 CDC 增量同步 已成功写入 7 条记录到表 order_agg注意最后一句DataY 发送的是7 条记录但 Doris 表里只会出现 6 行。合并发生在 Doris 的写入路径上。六、验证一全量快照也能直接汇总全量读完查 DorisSELECTstat_date,product_id,product_name,category,order_cnt,sales_amount,last_updateFROMcdc_demo.order_aggORDERBYstat_date,product_id;结果2026-10-25 1001 智能音箱 Pro 数码 2 798.00 2026-10-25 11:00:00 2026-10-25 1002 机械键盘 数码 2 598.00 2026-10-25 10:05:00 2026-10-25 1003 手冲咖啡壶 家居 3 267.00 2026-10-25 12:00:00 2026-10-26 1001 智能音箱 Pro 数码 2 798.00 2026-10-26 09:00:00 2026-10-26 1002 机械键盘 数码 1 299.00 2026-10-26 09:30:00 2026-10-26 1003 手冲咖啡壶 家居 2 178.00 2026-10-26 10:00:00对比源端 7 行2026-10-25 1001的两次记录合并成一行order_cnt从 1 1 变成 2sales_amount从 399 399 变成 798last_update用REPLACE保留了更晚的 11:00总行数从 7 变成 6。聚合结果agg_rows total_qty total_amount 6 12 2938.00源端同样按stat_date, product_id聚合结果完全一致。七、验证二Binlog 增量进来自动累加到已有 Key接下来模拟实时订单往 MySQL 再插 3 行INSERTINTOorder_detail(stat_date,product_id,product_name,category,order_cnt,sales_amount,last_update)VALUES(2026-10-25,1001,智能音箱 Pro,数码,3,1197.00,2026-10-25 18:00:00),(2026-10-26,1001,智能音箱 Pro,数码,1,399.00,2026-10-26 20:00:00),(2026-10-27,1004,降噪耳机,数码,2,998.00,2026-10-27 10:00:00);几秒钟后DataY 日志出现已成功写入 3 条记录到表 order_agg 组件DorisStreamLoad状态保存成功 {mysql_cdc.binlogFilemysql-bin.000003, mysql_cdc.binlogPosition6711}再次查询 Doris2026-10-25 1001 智能音箱 Pro 数码 5 1995.00 2026-10-25 18:00:00 2026-10-26 1001 智能音箱 Pro 数码 3 1197.00 2026-10-26 20:00:00 2026-10-27 1004 降噪耳机 数码 2 998.00 2026-10-27 10:00:00发生了三件事2026-10-25 1001从2 件 / 798 元累加为5 件 / 1995 元2026-10-26 1001从2 件 / 798 元累加为3 件 / 1197 元2026-10-27 1004是新的 Key直接新增一行。源端此时 10 行Doris 里 7 行source: 10 rows, qty 18, amount 5532.00 doris : 7 rows, qty 18, amount 5532.00增量路径同样没有丢数也没有重复累加。八、验证三1000 行批量写入仍然只留 10 个 Key最后做一次批量压力测试一次事务写入 1000 行覆盖 10 个商品。source_rows source_qty source_amount 1010 1018 105432.00DataY 侧已成功写入 1000 条记录到表 order_aggDoris 侧agg_rows total_qty total_amount 17 1018 105432.001000 行明细只产生 10 个聚合 Key每个商品正好 100 行2026-10-27 2001 批量商品2001 100 9990.00 2026-10-27 2002 批量商品2002 100 9990.00 ... 2026-10-27 2010 批量商品2010 100 9990.00到这里三条路径都验证完了场景源端行数Doris 行数件数金额全量快照76122938.00增量 3 行107185532.00批量 1000 行1010171018105432.00每一阶段Doris 的SUM结果都与源端GROUP BY完全一致。九、这次验证踩到的三个细节验证过程并不是一次就跑通的有三个细节值得记录。1. 目标表列名要对齐第一次建表时源表叫qty、amount、updated_at目标表叫order_cnt、sales_amount、last_update。结果 Key 列和维度列都进去了但指标列全是 0。原因很简单Stream Load 按 JSON 字段名找目标列名字对不上就取默认值。解决方式有两种源表和目标表使用一致的列名或者在中间加一个StreamSQL做字段改名。实时链路里建议把这一步固化成建表规范不要靠“应该能对上”。2. 聚合模型适合“追加”不适合“整行更新”这次验证的是订单流水每次销售都是新增一行天然适合SUM。如果源表会发生UPDATE比如订单金额被修改那么再把整行写进聚合模型就会重复累加原值 100UPDATE 后还是 100 聚合模型再 SUM 一次 → 200这类场景应该用 Doris 的Unique Key 模型做整行覆盖或者把更新转换成“冲正 新增”两条流水再进入聚合模型。模型选型比同步工具更重要。聚合模型不是万能的它适合的是“只追加、按 Key 汇总”的事实流。十、什么时候该用这套组合比较适合订单、支付、埋点、日志等追加型事实流看板只关心固定维度的SUM/COUNT/MAX/MIN明细量大但聚合后的 Key 数量可控希望省掉定时聚合任务降低延迟和存储。不太适合需要保留每一行明细源端频繁UPDATE/DELETE聚合 Key 基数接近明细行数那样聚合没有意义需要精确回溯某一条原始记录。十一、写在最后这次验证的核心结论只有一句话MySQL CDC 负责“把变化送过来”Doris 聚合模型负责“在落库时把账算好”。DataY 在这一链路里做的是三件事用MySQLBinlogInput做全量快照 Binlog 增量断点可恢复用DorisStreamLoad走 Stream Load 协议批量写入写入前按目标列类型做规范化保证字符、日期、数值不变形。最终效果1010 行明细 → 17 行汇总件数、金额与源端逐项对齐查询侧不用再跑GROUP BY看板直接读汇总表少一份明细存储少一套聚合任务少一段 T1 延迟。如果你的实时看板也在“先同步明细、再定时汇总”可以试试把汇总前移到写入阶段。很多时候链路不是不够长而是该在入口算的账被拖到了出口。十二、关于 DataY上面这条链路里的MySQLBinlogInput和DorisStreamLoad都来自 DataY。DataY 是一套开源、轻量的一体化数据平台理念很简单数据平台可以很小功能不打折。传统数据平台往往需要 Hadoop、Spark、Flink、ClickHouse 一整套组件几十台机器起步运维门槛高。DataY 选择另一条路一个 JAR自研流批一体处理引擎拒绝组件地狱2 核 4G就能跑通数据集成 → 数据建模 → 数据开发 → 任务调度 → 数据应用零依赖启动10 分钟搭好开发环境调度 master 与执行 worker 可分可合单机和集群同一套代码内嵌 DuckDB 作为高性能分析引擎同时支持 Doris、ClickHouse 等外部分析引擎。这里的“轻量”指的是运行资源门槛低不是能力轻量。DataY 由两块组成模块定位说明DataY Core数据集成引擎轻量、可嵌入、可扩展、流批一体JSON 定义任务、自动建表、CDC 实时同步可脱离 Web 独立运行或嵌入 Java 应用DataY Web可视化数据平台拖拽式任务设计、任务调度、维度建模、指标管理、智能问数和本文这条实时链路相关的能力多数据源MySQL / PostgreSQL / Oracle / SQL Server / SAP HANA / Doris / ClickHouse / DuckDB 等二十余种数据源四种 CDCMySQL Binlog、PostgreSQL WAL、Oracle LogMiner、MongoDB Change Streams统一组件模型Source / Operator / Sink批流共用️断点续传Binlog 位点、LSN、SCN、resumeToken 统一持久化️自动建表跨库 DDL 转换表不存在时自动创建Doris 流式写入内置DorisStreamLoad支持 insert / delete / overwrite流内计算StreamSQL、Join、Lookup、DataTransform同步路上顺手把数据加工好⏰任务调度Quartz Cron 精确到秒DAG 拓扑编排master/worker 可分离MCP Harness把数据源、数据服务、ETL 任务、数据模型暴露给 AI Agent让 Agent 直接操作平台对象可嵌入Core 可以作为 Java 库嵌进你自己的应用。快速开始gitclone https://gitee.com/yuyuejia/datay.gitcddatay mvn clean package-DskipTestsjava-jardatay-web/target/*.jar--DATAY_AI_API_KEYsk-xxxxxxxxx# 浏览器打开 http://localhost:8080默认账号 admin/admin也可以只用 Core 跑单个任务java-jardatay-core/target/datay-core-*-jar-with-dependencies.jar taskConfig.json产品首页http://datay.yuyuejia.com.cn/在线体验http://datay-demo.yuyuejia.com.cn/项目地址https://gitee.com/yuyuejia/datayDataY 想做的不是再堆一套大数据组件而是把数据集成、建模、调度、数仓和问数收进一个足够轻的平台上。验证环境DataY 1.0.4 / MySQL 8.0.46 / Apache Doris 2.1.0 / MySQL Connector/J 8.0.33。
