数据流水线有一个老大难问题上游来的数据质量不归你管但出了事算你的。凌晨三点定时 ETL 把昨天的一批新数据灌进生产表events——报表、看板、下游作业、特征管道全都在读它。这批数据里混着上游常见的脏东西空user_id、负数金额、一眼假的离群值、对不上维表的用户。等白天有人发现时晨会看板已经算错、下游作业已经跑完、模型已经拿它训了一轮。然后是更难的部分脏数据和好数据已经混在同一张表里事后把它摘干净比当初挡在门外难十倍——那又回到了第五篇的事故救援。软件工程对这类问题的标准答案是CI 门禁代码得先过测试才能合进主干。数据世界的对应物叫Write-Audit-PublishWAP——新数据先写进隔离区、通过审计、再发布。过去要搭这套得靠 Iceberg / lakeFS 这类湖上工具在MatrixOne里靠内建的 git4data 能力它就是分支的一个基本用法。这一篇我们把它写细什么时候需要它、三步怎么落地、以及别的方案为什么不好使。SQL 全部在 MatrixOne4.0.0-rc3上实测过。 本文 SQL 整体可跑matrixorigin/git4data-tutorial 的07-write-audit-publish/。环境docker run -d -p 6001:6001 --name matrixone matrixorigin/matrixone:4.0.0-rc3。什么时候你会需要 WAPWAP 不是每张表都要上——它专治的是ETL数据管道里的一类处境一张有下游消费者的表被管道定期灌入来自你不完全信任的来源的新数据。下面这些都是 ETL 场景本质是同一件事——你在跑一条数据管道往一张被人读的表里灌数每天 / 每小时的 ETL 批量入库一张被报表、看板、下游作业持续读的表凌晨批量灌数。一次坏批次早上整个公司看到的都是错的。接入你不掌控的外部数据源合作方的数据 feed、第三方 API、爬来的数据、用户上传——质量参差今天好好的明天上游改了个字段就全乱。给训练 / 特征管道供数脏数据不会报错它只会静默地把模型训歪上一篇那张特征表最怕这个。发布一张很多人依赖的口径表 / 指标表一处口径错下游一片跟着错。回流数据到业务系统Reverse ETL把算好的结果推回生产库、推给营销 / 风控系统——一次错误推送是有真实后果的。这些 ETL 场景的共同点问题数据一旦进了生产表伤害就已经发生了下游读到了、决策做了、模型训了再去补救就是事故。WAP 的整个价值就是把发现问题这件事从生产之后提前到发布之前——相当于给你的 ETL 管道加一道发布前的门禁。WAP 三步Write → Audit → Publish下面用一张真实语境的表把三步走完。生产表events有 10 万行昨天的干净数据下游持续在读另有一张维表dim_users事实表的user_id必须能对上它。先把这两张表建好、灌入昨天的干净数据——这段是可复现的前置CREATEDATABASEwap_demo;USEwap_demo;-- 生产事实表昨天的 10 万行干净数据下游报表 / 看板 / 下游作业持续在读CREATETABLEevents(event_idBIGINTPRIMARYKEY,user_idINT,amountDECIMAL(10,2),statusVARCHAR(16),tsDATE);INSERTINTOeventsSELECTresult,result%5000,round(rand()*5001,2),paid,2026-06-29FROMgenerate_series(1,100000)g;-- 维表事实表的 user_id 必须能在它里面找到CREATETABLEdim_users(user_idINTPRIMARYKEY,tierVARCHAR(8));INSERTINTOdim_usersSELECTresult,stdFROMgenerate_series(0,4999)g;Write新数据永远先落在 staging 分支今天的新批次绝不直接碰生产表。先给它开一条 staging 分支——毫秒级、零拷贝第三篇讲过原理再把批次灌进这条分支DATABRANCHCREATETABLEevents_stageFROMevents;-- staging 分支毫秒级-- 今天的批次 5000 行落到 staging。里面混着真实上游常见的脏数据-- 空 user_id、对不上维表的用户、负数金额、离谱离群值INSERTINTOevents_stageSELECT200000result,CASEWHENresult%970THENNULLWHENresult%890THEN999999-- 维表里没有的用户ELSEresult%5000END,CASEWHENresult%2500THEN-1.00WHENresult%3330THEN999999.99ELSEround(rand()*5001,2)END,paid,2026-06-30FROMgenerate_series(1,5000)g;此刻生产表events一行未动下游读到的还是昨天那份干净数据。脏数据被关在 staging 里——这就是 WAP 的第一性原理先隔离再谈质量。AuditSQL 就是质量门禁审计就是一组跑在 staging 上的 SQL 断言——每一条都应该返回 0只要有一条不为 0门禁就不放行。一套像样的审计通常覆盖这几类检查-- ① 完整性 取值域 业务规则关键字段非空、金额合理、状态合法SELECTSUM(CASEWHENuser_idISNULLTHEN1ELSE0END)ASnull_user,SUM(CASEWHENamount0THEN1ELSE0END)ASnegative_amount,SUM(CASEWHENamount10000THEN1ELSE0END)ASoutlier_amount,SUM(CASEWHENstatusNOTIN(paid,refunded,void)THEN1ELSE0END)ASbad_statusFROMevents_stageWHEREts2026-06-30;-- ② 参照完整性批次里的 user_id 必须都能在维表里找到SELECTCOUNT(*)ASorphan_usersFROMevents_stage sLEFTJOINdim_users dONs.user_idd.user_idWHEREs.ts2026-06-30ANDs.user_idISNOTNULLANDd.user_idISNULL;-- ③ 体量今天批次的行数要落在合理区间防上游双灌、防空跑SELECTCOUNT(*)ASbatch_rowsFROMevents_stageWHEREts2026-06-30;实测这批脏数据门禁一条条把问题抓了出来检查结果null_user空用户51✗negative_amount负数金额20✗outlier_amount离群值15✗orphan_users对不上维表56✗batch_rows体量5000 ✓门禁不放行。这里还能顺手加更多检查——唯一性主键 / 自然键无重复GROUP BY key HAVING COUNT(*) 1、新鲜度MAX(ts)得是今天、分布漂移这批的空值率 / 金额均值和历史比有没有突变。它们都只是 SQL想加就加。门禁不过怎么办直接拒收生产零感知关键就在这一步门禁没过你什么都不用对生产做——因为生产从头到尾就没碰过这批数据。把 staging 分支一扔即可DROPTABLEevents_stage;-- 拒收这一批SELECTCOUNT(*)FROMevents;-- 还是 100000一行没动SELECTCOUNT(*)FROMeventsWHEREts2026-06-30;-- 0这批根本没进来实测拒收之后生产表还是10 万行今天这批脏数据一行都没进生产。接下来就是排查上游、修好、重跑——而不是在生产上做事故救援。想留现场 debug别 DROP留着这条 staging 分支就行——它就是一份完整、隔离的事故现场你可以慢慢查上游到底发了什么期间完全不影响生产。Publish通过审计一次原子发布排查修好上游后换一批干净数据重跑——同样先落 staging 分支DATABRANCHCREATETABLEevents_stageFROMevents;INSERTINTOevents_stageSELECT200000result,result%5000,round(rand()*5001,2),paid,2026-06-30FROMgenerate_series(1,5000)g;这次审计全绿null_user / negative / outlier / orphan全 0、MAX(ts)是今天。发布前再用 DIFF 看一眼这批究竟会给生产带来什么——行级、确切DATABRANCH DIFF events_stage AGAINST events OUTPUT SUMMARY;-- INSERTED 5000确切就是这 5000 行确认无误一次原子合并DATABRANCHMERGEevents_stageINTOevents;这一步是原子的下游读者要么看到完整的、已审计的整批 5000 行要么在这条语句之前一行也看不到——不存在发布到一半的中间状态。实测SELECTCOUNT(*)FROMevents;-- 105000SELECTCOUNT(*)FROMeventsWHEREuser_idISNULLORamount0ORamount10000;-- 0生产表从 10 万变 10.5 万而且从头到尾没出现过一行脏数据。别的方案怎么做带上数据库选型逐个对比“不就是灌数前先检查一下嘛” 最朴素的做法——直接灌进生产、事后再查——也最常见、最痛等检查发现问题脏数据早已在生产表里、被下游读走了这就是先发布再祈祷。WAP 就是为了避免这一幕。但真要落地 WAP做法取决于你手上是什么系统。分两类看一、湖上的 git 式方案Iceberg 分支、lakeFSWAP 在这里被做成了原生能力Apache Iceberg分支写进一条 audit 分支spark.wap.branch、在分支上跑质量检查、通过就把mainfast-forward 到 audit 分支fast_forward(t,main,audit)让主分支追上。分支零拷贝、fast-forward 是纯元数据操作且原子——思路和 MatrixOne 的做法几乎一样。差异在落点Iceberg 是对象存储上的表格式自己不执行查询得靠 Spark / Trino / Flink 这类外部引擎来读写、跑审计它面向分析AP生产表是湖上给分析读的数据集不是一个还在对外服务点查 / 事务的库。要跑 WAP得配 engineSpark 的 WAP 模式、catalog 和一整套 lake 栈。lakeFS给整个数据湖做 git——branch 整个 repo、写到分支、用pre-merge hooks_lakefs_actions/下的 action 文件当审计门禁、通过才 merge 到main。merge 是 repo 级原子天然多表 / 多文件一致。差异lakeFS 版本化的是对象存储里的文件 / 路径不是数据库表它挡在 S3 前面你仍要用外部引擎Spark / Trino / DuckDB在版本化路径上查数审计逻辑跑在 webhook / Airflow 里另一套编排。这两者和 MatrixOne本质上是同一条路——都能完整地实现 WAP。区别在落点它们保护的是湖 / 数仓里给分析读的数据集需要存储格式 外部引擎 catalog / hooks一整套栈而 MatrixOne 把同一套 git 式 WAP 直接做进它自己这个还在对外服务的 HTAP 数据库——审计是同库里的普通 SQL发布是库内的原子 MERGE读的人就是这个库的消费者。二、传统数仓 / 数据库上凑WAP没有 git 分支只能靠交换Snowflake零拷贝CLONE出 staging、灌 审计再ALTER TABLE prod SWAP WITH staging两条 RENAME 在一个事务里、原子注意权限不随数据走staging 表要COPY GRANTS才继承 prod 的授权。clone 零拷贝这点很像分支。差异SWAP 换掉的是整张表、不是增量发布想只追加直接INSERT/MERGE也行但那就绕过了“先隔离再发布”这道门、回到方案 A多表要原子得自己拼且标准表面向分析AP要低延迟点查 / OLTP 得另用 Hybrid TablesUnistore。PostgreSQL / MySQL没有分支。可行的近似是分区交换——把新批次灌进一张单独的表、校验再ALTER TABLE … ATTACH PARTITIONPG/EXCHANGE PARTITIONMySQL挂上去或者 staging 表 RENAME交换、外面用事务包住。差异分区交换是把一张已填好数据的表挂载成一个分区RANGE / LIST / HASH 都行数据要符合该分区约束是分区粒度的挂载 / 替换、不是行级追加或更新挂载要拿锁、还要过 CHECK 约束扫描RENAME 交换有前面说的多表不原子、句柄 / 索引 / 权限重建问题事务包住则长事务持锁、拖垮在线表。BigQuery没有分支。staging 表 事务里MERGE/CREATE OR REPLACE TABLE/ 分区覆写或表快照 覆盖。形态同上要么整表 / 整分区替换要么大 MERGE 有中间态和成本。这类系统都在绕过没有分支这件事——用整表 / 整分区的交换来近似原子发布。代价是只能整块换不适合行级增量 upsert、多表难原子、换的过程要么锁表、要么重建一堆附属对象。方案隔离机制发布机制增量追加多表原子需外部引擎能服务在线读Iceberg 分支零拷贝分支fast-forward元数据原子支持弱表级是Spark/Trino否湖上 APlakeFSrepo 分支merge原子带 hooks 门禁支持强repo 级是否文件层Snowflake零拷贝 cloneSWAP整表交换原子整表换弱否自带否标准表Hybrid 例外PG / MySQLstaging 表 / 分区ATTACH / EXCHANGE / RENAME按分区弱否是但要锁 / 重建MatrixOnegit4data 能力零拷贝分支原子 MERGE行级增量原生库级快照兜底否同引擎 SQL是HTAP直接服务一句话总结WAP 这套 git 式门禁过去要么在湖上做Iceberg / lakeFS——但那不是能对外服务的库还得配一套引擎栈要么在数仓 / 库上用整表交换硬凑Snowflake SWAP、PG 分区交换——粒度粗、多表难原子。MatrixOne 是少见的把它做进一个还在对外服务的 HTAP 数据库、且发布是行级增量原子 MERGE的方案——审计就是同一个库里的 SQL不用再搭一套。多表要一起发布用库级快照兜底如果一次发布要同时更新事实表和多张维表想要要么全发、要么全不发可以在发布前给整个库打一个快照再逐表MERGE任何一步审计 / 合并出问题就RESTORE DATABASE db {SNAPSHOT s}把整库一起拨回发布前——多表的原子性用第五篇讲过的库级快照来兜底单条DATA BRANCH MERGE本身是表级的。把它接进流水线数据的 CI/CD这套流程天生适合自动化挂到调度器 / CI 里每天的批次自动走一遍Write批次到达 → 自动DATA BRANCH CREATE一条当天的 staging 分支灌入。Audit自动跑那组审计 SQL任一条 0 → 判定失败。Publish / Reject全绿 →DATA BRANCH MERGE发布有红 →不发布、告警、并保留 staging 分支当现场。心智模型的变化才是关键没有 WAP生产表 数据的入口质量问题进来了再说。有了 WAP生产表 数据的出口只有通过审计的数据才配进来。这就是数据的 CI/CD门禁不过坏数据连生产的门都进不来。成本与边界门禁几乎免费staging 分支毫秒级、零拷贝审计就是普通 SQL发布是一次秒级 MERGE与表多大无关。这道门禁不会成为流水线的瓶颈。审计的强度 你写的检查的强度WAP 给的是能可靠地拦住这个机制拦什么得你用 SQL 定义。机制再好也替不了你想清楚这批数据怎样才算合格。分支 / 快照会占存储直到释放被 staging 分支或快照钉住的历史对象不会被后台 GC拒收后记得DROP留作现场的也要有清理策略。多表原子发布靠库级快照兜底见上。行级 diff / merge 要求两边 schema 一致第四篇的边界批次要新增列先在主线改 schema再灌值。结语至此数据运维三部曲收官个人的安全网第五篇出事能秒级回退、团队的并行第六篇分支与合并、生产的门禁本篇脏数据进不来。三件事共用同一套原语——snapshot、branch、diff、merge——这正是把版本控制装进数据库的意义不是多了一个功能而是多了一种工作方式。下一篇起进入AI 训练主题。第一站是最经典的一个问题数据每天都在变凭什么每次都全量重训用 DIFF 把变了的那部分精确取出来只训增量。 可运行 SQLgithub.com/matrixorigin/git4data-tutorial 源码与社区github.com/matrixorigin/matrixone