鲁班 - 系统设计大全

60.2 数仓分层 (ODS / DWD / DWS / ADS)

一句话

数仓分层不是「多写几遍 SQL」，而是把「清洗一次、复用 N 次」变成物理约束：ODS 只贴源、DWD 一单一行、DWS 只做聚合、ADS 只做口径；每一层都能从上游重算，才敢删、敢改、敢补数。

六个误区

误区 现实
分层 = 多建几张表 分层是粒度与职责的切分；表多不等于分层，表少也不等于没分层
ODS 可以修数据、补字段 ODS 必须可重放；任何修改都让「重跑一遍」对不上账
DWD 越宽越好 宽表属于 DWS / ADS；DWD 要保持业务过程粒度，否则一行的含义就说不清
DWS 可以跳过，直接 DWD 出报表 小团队可以省一层，但必须有「单一事实来源」；否则每个报表各写一遍聚合
ADS 可以直接查业务库 报表绕过数仓，口径不可控，业务库还被分析查询拖慢
分层后查询一定更快 分层增加跳数；服务层要靠物化、列存、预聚合，不是靠「层数」

四层各管什么

层 一行代表什么 允许的操作 典型分区 保留 消费者
ODS 源系统的一次投递（可重复） 只追加、只做接入校验 按 ingested_at / dt 30–90 天 DWD 任务
DWD 一个业务过程的最小粒度（一单 / 一事件 / 一日快照） 清洗、去重、规范化、维度关联 按业务日期 dt 1–3 年 DWS、明细查询
DWS 一个主题 + 一个粒度（日 / 品类 / 用户） 聚合、窗口、累计 按统计日期 dt 1–2 年 ADS、分析
ADS 一个应用指标 / 一张宽表 口径映射、格式转换、轻量计算 按应用日期 按合规 BI、API、特征

命名把层级、域、实体、粒度一次说清：

{layer}_{domain}_{entity}_{granularity}
dwd_trade_order_di      -- di：daily increment，按天增量
dws_trade_sales_1d      -- 1d：按天汇总
ads_trade_gmv_1d        -- 面向应用的 GMV 日指标
dim_user_df             -- df：daily full，每日全量快照

🔑 分层的核心产物不是表，而是粒度声明：dwd_trade_order_di 一行必须是一个订单的最新状态，dws_trade_sales_1d 一行必须是一个「日期」的汇总。粒度说不清，去重、聚合、对账全都会错。

四条原理，缺一条分层就塌

① 单一事实来源：口径只定义一次

同一个「GMV」如果有三个定义（含运费 / 不含退款 / 含取消），它就不是一个指标。分层的做法是：在 DWD 固定事实与状态，在 ADS 固定口径，中间层不再重新解释业务。

口径 定义位置 例子
事实 DWD dwd_trade_order_di：订单金额、状态、支付时间
汇总 DWS dws_trade_sales_1d：按日汇总已支付 + 已退款
指标 ADS ads_trade_gmv_1d：GMV = 已支付 + 已退款；净 GMV = GMV − 退款

② 可重算：每一层都是上游的确定性函数

只要给定相同的上游分区和相同的代码版本，重跑必须得到相同结果。做不到这一点，分层就只是「把数据抄了一遍」，补数、修 bug、迁移引擎全都不敢动。

可重算的两个前提：

  1. 输入确定：读固定分区，不读 now()、不读「最新状态」这种漂移输入；
  2. 输出可覆盖：按分区整体替换，而不是「追加了事」。

③ 幂等：重跑 N 次 = 跑 1 次

幂等不是「加个 DISTINCT」，而是三件事同时成立：

要素 做法 反例
确定性 排序、聚合、窗口都有确定语义 ROW_NUMBER() 不指定排序，每次结果不同
可覆盖 INSERT OVERWRITE / 先删分区再插 直接 INSERT INTO，重跑翻倍
去重键 业务主键 + 更新时间 用自增 ID 去重，重放后 ID 全变

④ 时间语义：事件时间、处理时间、迟到数据

时间 含义 用在哪
事件时间 业务发生时间（下单、支付） 分区、窗口、报表口径
处理时间 任务实际运行时间 调度、监控、SLA
摄入时间 数据进入 ODS 的时间 追溯「这批数据什么时候到」

迟到数据（退款、改单、补录）会改写历史分区。所以「只重算今天」是错的，必须按受影响的事件日期重算；影响的日期范围由业务决定（退款可能回溯 30 天）。

⚠️ 最常见的账错：ODS 里同一订单有「创建」「支付」「退款」多条记录，报表直接 SUM(amount) WHERE status='paid'，把历史支付和当前退款混在一起算。DWD 的职责就是把它收敛成「一单一行、最新状态」。

代码：四层跑一遍，看直查 ODS 到底错多少

#!/usr/bin/env python3
"""最小可跑数仓：ODS -> DWD -> DWS -> ADS（SQLite，无第三方依赖）。"""
import hashlib
import sqlite3

# 同一批订单被投递了两次，o2 金额被修正；第三批是迟到的退款
BATCH1 = [
    ("o1", "u1", 10000, "paid",     "2026-10-01T10:00:00", "2026-10-01"),
    ("o2", "u2", 20000, "paid",     "2026-10-01T11:00:00", "2026-10-01"),
    ("o3", "u1",  5000, "canceled", "2026-10-02T09:00:00", "2026-10-02"),
    ("o4", "u3", 30000, "paid",     "2026-10-02T12:00:00", "2026-10-02"),
    ("o5", "u2", 15000, "paid",     "2026-10-03T08:00:00", "2026-10-03"),
    ("o6", "u1", 12000, "paid",     "2026-10-03T15:00:00", "2026-10-03"),
    ("o7", "u3",  8000, "paid",     "2026-10-03T18:00:00", "2026-10-03"),
    ("o8", "u2",  9000, "created",  "2026-10-04T07:00:00", "2026-10-04"),
]
BATCH2 = [(o, u, (21000 if o == "o2" else a), s,
           ("2026-10-05T01:04:00" if o == "o2" else t), d)
          for o, u, a, s, t, d in BATCH1]
BATCH3 = [
    ("o2", "u2", 21000, "refunded", "2026-10-06T01:58:00", "2026-10-01"),
    ("o5", "u2", 15000, "refunded", "2026-10-06T01:59:00", "2026-10-03"),
    ("o9", "u3",  6000, "paid",     "2026-10-04T20:00:00", "2026-10-04"),
]

DDL = """
CREATE TABLE ods_orders (            -- 贴源：不设主键、不更新、只追加
  order_id TEXT, user_id TEXT, amount_cents INTEGER, status TEXT,
  updated_at TEXT, dt TEXT, batch_id TEXT
);
CREATE TABLE dwd_trade_order_di (    -- 明细：一单一行，保留最新状态
  order_id TEXT PRIMARY KEY, user_id TEXT, amount_cents INTEGER,
  status TEXT, updated_at TEXT, dt TEXT
);
CREATE TABLE dws_trade_sales_1d (    -- 轻度汇总：按订单日期
  dt TEXT PRIMARY KEY, paid_orders INTEGER, gmv_cents INTEGER,
  refund_cents INTEGER, net_cents INTEGER
);
CREATE TABLE ads_trade_gmv_1d (      -- 应用层：只给看板/接口用
  dt TEXT PRIMARY KEY, gmv_yuan REAL, net_yuan REAL, refund_rate REAL
);
"""


def ingest(con, rows, batch_id):
    con.executemany(
        "INSERT INTO ods_orders VALUES (?,?,?,?,?,?,?)",
        [(*r, batch_id) for r in rows],
    )


def build_dwd(con):
    """幂等合并：同一 order_id 只留 updated_at 最大的一行。"""
    con.executescript("""
    WITH latest AS (
      SELECT *, ROW_NUMBER() OVER (
               PARTITION BY order_id ORDER BY updated_at DESC, batch_id DESC
             ) AS rn
      FROM ods_orders
    )
    INSERT INTO dwd_trade_order_di
    SELECT order_id, user_id, amount_cents, status, updated_at, dt
    FROM latest WHERE rn = 1
    ON CONFLICT(order_id) DO UPDATE SET
      amount_cents = excluded.amount_cents,
      status       = excluded.status,
      updated_at   = excluded.updated_at,
      dt           = excluded.dt
    WHERE excluded.updated_at > dwd_trade_order_di.updated_at;
    """)


def build_marts(con, dts):
    """按分区重建：先删后插 + 事务，重跑结果不变。"""
    con.execute("BEGIN")
    for dt in dts:
        con.execute("DELETE FROM dws_trade_sales_1d WHERE dt = ?", (dt,))
        con.execute("""
        INSERT INTO dws_trade_sales_1d
        SELECT dt,
               SUM(CASE WHEN status = 'paid' THEN 1 ELSE 0 END),
               SUM(CASE WHEN status IN ('paid','refunded') THEN amount_cents ELSE 0 END),
               SUM(CASE WHEN status = 'refunded' THEN amount_cents ELSE 0 END),
               SUM(CASE WHEN status IN ('paid','refunded') THEN amount_cents ELSE 0 END)
                 - SUM(CASE WHEN status = 'refunded' THEN amount_cents ELSE 0 END)
        FROM dwd_trade_order_di
        WHERE dt = ? AND status IN ('paid','refunded') GROUP BY dt
        """, (dt,))
        con.execute("DELETE FROM ads_trade_gmv_1d WHERE dt = ?", (dt,))
        con.execute("""
        INSERT INTO ads_trade_gmv_1d
        SELECT dt, gmv_cents / 100.0, net_cents / 100.0,
               CASE WHEN gmv_cents = 0 THEN 0
                    ELSE ROUND(refund_cents * 1.0 / gmv_cents, 4) END
        FROM dws_trade_sales_1d WHERE dt = ?
        """, (dt,))
    con.commit()


def checksum(con, table, cols):
    rows = con.execute(f"SELECT {cols} FROM {table} ORDER BY 1").fetchall()
    return hashlib.sha256(repr(rows).encode()).hexdigest()[:12]


def main():
    con = sqlite3.connect(":memory:")
    con.executescript(DDL)
    ingest(con, BATCH1, "2026-10-05T01:00")
    ingest(con, BATCH2, "2026-10-05T01:05")
    print("ODS 原始行数：", con.execute("SELECT COUNT(*) FROM ods_orders").fetchone()[0],
          "（两批投递，其中 8 行是重复）")

    print("\n① 报表直接查 ODS（错误示范）")
    print("  rows_cnt | gmv_yuan")
    for row in con.execute("""SELECT COUNT(*), ROUND(SUM(amount_cents)/100.0,2)
                              FROM ods_orders WHERE status='paid'"""):
        print("  " + " | ".join(map(str, row)))

    build_dwd(con)
    build_marts(con, ["2026-10-01", "2026-10-02", "2026-10-03", "2026-10-04"])
    print("\n② DWD 明细（一单一行）")
    for row in con.execute("""SELECT order_id, amount_cents/100.0, status, dt
                              FROM dwd_trade_order_di ORDER BY order_id"""):
        print("  " + " | ".join(map(str, row)))
    print("\n③ DWS 日汇总")
    for row in con.execute("""SELECT dt, paid_orders, gmv_cents/100.0, refund_cents/100.0,
                                     net_cents/100.0 FROM dws_trade_sales_1d ORDER BY dt"""):
        print("  " + " | ".join(map(str, row)))
    print("\n④ ADS 应用层")
    for row in con.execute("""SELECT dt, gmv_yuan, net_yuan, refund_rate
                              FROM ads_trade_gmv_1d ORDER BY dt"""):
        print("  " + " | ".join(map(str, row)))

    h1 = (checksum(con, "dws_trade_sales_1d", "*"), checksum(con, "ads_trade_gmv_1d", "*"))
    build_dwd(con)
    build_marts(con, ["2026-10-01", "2026-10-02", "2026-10-03", "2026-10-04"])
    h2 = (checksum(con, "dws_trade_sales_1d", "*"), checksum(con, "ads_trade_gmv_1d", "*"))
    print(f"\n⑤ 幂等重跑：{h1[0]}/{h1[1]} vs {h2[0]}/{h2[1]} -> "
          f"{'一致' if h1 == h2 else '不一致'}")

    ingest(con, BATCH3, "2026-10-06T02:00")
    build_dwd(con)
    build_marts(con, ["2026-10-01", "2026-10-03", "2026-10-04"])
    print("\n⑥ 迟到退款后（只重算 10-01 / 10-03 / 10-04）")
    for row in con.execute("""SELECT dt, gmv_yuan, net_yuan, refund_rate
                              FROM ads_trade_gmv_1d ORDER BY dt"""):
        print("  " + " | ".join(map(str, row)))

    direct = con.execute("SELECT SUM(amount_cents)/100.0 FROM ods_orders WHERE status='paid'").fetchone()[0]
    gross = con.execute("SELECT SUM(gmv_yuan) FROM ads_trade_gmv_1d").fetchone()[0]
    net = con.execute("SELECT SUM(net_yuan) FROM ads_trade_gmv_1d").fetchone()[0]
    print(f"\n最终对账：ODS 直查 {direct:.2f} 元；ADS 毛 {gross:.2f} 元 / 净 {net:.2f} 元；"
          f"直查多算 {(direct-gross)/gross*100:.1f}%")


main()

本机实测（macOS，Apple M1，Python 3.14.8）：

ODS 原始行数： 16 （两批投递，其中 8 行是重复）

① 报表直接查 ODS（错误示范）
  rows_cnt | gmv_yuan
  12 | 1910.0

② DWD 明细（一单一行）
  o1 | 100.0 | paid | 2026-10-01
  o2 | 210.0 | paid | 2026-10-01
  o3 | 50.0 | canceled | 2026-10-02
  o4 | 300.0 | paid | 2026-10-02
  o5 | 150.0 | paid | 2026-10-03
  o6 | 120.0 | paid | 2026-10-03
  o7 | 80.0 | paid | 2026-10-03
  o8 | 90.0 | created | 2026-10-04

③ DWS 日汇总
  2026-10-01 | 2 | 310.0 | 0.0 | 310.0
  2026-10-02 | 1 | 300.0 | 0.0 | 300.0
  2026-10-03 | 3 | 350.0 | 0.0 | 350.0

④ ADS 应用层
  2026-10-01 | 310.0 | 310.0 | 0.0
  2026-10-02 | 300.0 | 300.0 | 0.0
  2026-10-03 | 350.0 | 350.0 | 0.0

⑤ 幂等重跑：4ef3aa751722/7cb5233faf74 vs 4ef3aa751722/7cb5233faf74 -> 一致

⑥ 迟到退款后（只重算 10-01 / 10-03 / 10-04）
  2026-10-01 | 310.0 | 100.0 | 0.6774
  2026-10-02 | 300.0 | 300.0 | 0.0
  2026-10-03 | 350.0 | 200.0 | 0.4286
  2026-10-04 | 60.0 | 60.0 | 0.0

最终对账：ODS 直查 1970.00 元；ADS 毛 1020.00 元 / 净 660.00 元；直查多算 93.1%

读数：

现象 数字 原因
ODS 直查 GMV 1970.00 元 16 行里 12 行 paid：两批重复 + o2 的旧金额 + 退款单仍被当支付
ADS 毛 GMV 1020.00 元 DWD 去重后 9 单，其中 2 单退款仍计毛额
ADS 净 GMV 660.00 元 扣掉 210 + 150 两笔退款
虚高比例 93.1% 接近「翻倍」——重复投递与状态未收敛的复合效应
幂等校验 两次哈希相同 分区先删后插 + 事务，重跑不累加
迟到退款 只重算 3 个分区 影响面按事件日期算，不是「只跑今天」

🔑 这张表解释了分层最朴素的价值：ODS 是「发生过什么」，DWD 是「现在是什么」，DWS 是「按什么粒度看」，ADS 是「业务怎么定义」。四件事混在一条 SQL 里，数字就会像上面一样虚高 93%。

增量、全量与拉链：三种更新策略

策略 做法 适用 代价
全量覆盖 每次重算整个分区 / 整表 维表、小表、逻辑频繁变 计算成本随历史增长
增量合并 只处理新分区，按主键 MERGE 事实表、日志、订单 需要可靠的主键与更新时间；迟到数据要回刷
拉链表（SCD-2） 关闭旧版本 + 插入新版本，保留生效区间 会变且要回溯的维度（类目、等级、归属） 查询要带时间条件；写入要事务

拉链的 SQL 骨架（两条语句必须在同一事务里）：

-- 1) 关闭旧版本：只改「当前有效」的那一行
UPDATE dim_user
SET end_date = :biz_date, is_current = 0
WHERE user_id = :uid AND is_current = 1 AND start_date < :biz_date;

-- 2) 插入新版本
INSERT INTO dim_user(user_id, level, start_date, end_date, is_current)
VALUES (:uid, :new_level, :biz_date, '9999-12-31', 1);

⚠️ 拉链表最常见的两个 bug：先插后关导致同一时刻两行有效；关闭条件漏了 is_current=1，把历史版本全改成同一天结束。建模细节见 46.1 数据建模。 CDC 的三种操作（I / U / D）在 ODS 必须原样保留（通常加 op 字段），到 DWD 再按主键合并。源库的物理删除要转成软删（is_deleted / deleted_at），否则下游无法区分「这条记录今天没来」与「这条记录被删了」——前者是延迟，后者是业务事实，处理方式完全相反。CDC 的日志格式与快照对齐见 46.4 CDC (变更数据捕获)。

调度：依赖数据就绪，不是依赖时钟

依赖类型 写法 风险
时间依赖 每天 02:00 跑 上游晚到就空跑，产出「0 行」而不是失败
数据就绪 等上游分区 _SUCCESS / 行数 > 0 需要传感器与超时；上游永久不产出会一直等
事件驱动 上游完成后发消息触发 消息丢失要有兜底扫描

生产上三层一起用：数据就绪为主、时间为兜底、消息做加速。调度器语义见 61.3 批处理与调度。

补数（backfill）的三条纪律：

  1. 先算影响分区（按事件日期），再按分区串行或小并发重跑；
  2. 重跑前确认上游分区齐、代码版本与当时一致（或明确接受新逻辑）；
  3. 每跑完一个分区做一次行数/金额对账，不通过就停，不要一路跑到底。

生产落地：指名 → 可抄 → 可验 → 可退

指名：模型层用 dbt（SQL 版本化 + 测试 + 血缘）或 Spark SQL / Hive SQL（大数据量）；表格式用 Apache Iceberg（见 60.3 数据湖与湖仓一体）；查询用 Trino / Presto；调度用 Airflow（或 DolphinScheduler）。

可抄（dbt + Spark 适配器，未验证：本机未安装 dbt）：

-- models/marts/dws_trade_sales_1d.sql
{{ config(
    materialized='incremental',
    incremental_strategy='merge',
    unique_key='dt',
    partition_by={'field': 'dt', 'data_type': 'date'}
) }}

select dt,
       sum(case when status = 'paid' then 1 else 0 end) as paid_orders,
       sum(case when status in ('paid','refunded') then amount_cents else 0 end) as gmv_cents,
       sum(case when status = 'refunded' then amount_cents else 0 end) as refund_cents
from {{ ref('dwd_trade_order_di') }}
{% if is_incremental() %}
where dt >= date_sub(current_date(), 3)   -- 迟到数据回刷窗口
{% endif %}
group by dt

# 只跑这一层，再跑测试；dbt 会按 ref 自动排依赖
dbt run --select dws_trade_sales_1d
dbt test --select dws_trade_sales_1d
# 预期：Completed successfully；PASS=5 WARN=0 ERROR=0

可验：本节脚本的哈希一致性就是最小验收；生产上用「同一分区跑两次，EXCEPT 双向为空」验证幂等：

SELECT * FROM dws_trade_sales_1d WHERE dt = '2026-10-01'
EXCEPT
SELECT * FROM dws_trade_sales_1d_bak WHERE dt = '2026-10-01';

可退：每次发布保留上一版分区或 Iceberg 快照；逻辑错误时先回滚表到上一个快照，再修代码重跑，而不是直接改线上数据。小表可用 dbt run --full-refresh 重建；大表按分区回滚，避免全量重算。

反模式

反模式 后果
ODS 里做清洗、改字段 无法重放；重跑与历史对不上，审计断链
DWD 直接 SELECT * 不做去重 重复投递直接放大指标（本节实测虚高 93%）
用 ROW_NUMBER() 却不写 ORDER BY 每次跑取到不同行，幂等校验永远失败
DWS / ADS 用 INSERT INTO 追加 重跑一次翻一倍，补数后数字越补越大
只按处理时间分区 迟到数据无处可放，只能改「今天」，历史永远错
用视图代替 ADS 层 每次查询都重算，报表慢且拖垮集群
分层后不做层间对账 DWD 少一行、ADS 多一行，没人发现
指标口径写在报表 SQL 里 同一个 GMV 三个数，争议时拿不出证据

自检

Q1 ODS 为什么必须「只贴源、不修改」？如果源数据有错，应该在哪一层修？

▸ 答案：ODS 的职责是保留「系统实际收到过什么」，它是重放与审计的基线。一旦在 ODS 改数，就再也无法回答「当时到底收到了什么」，重跑也会与历史不一致。源数据错误应在 DWD 修：保留原始行，在 DWD 做清洗、映射、纠错，并记录修正规则与版本；这样既能重算，也能追溯。

Q2 DWD 的「一单一行」靠什么保证？去重键和排序键怎么选？

▸ 答案：靠「业务主键 + 版本字段」的合并：主键是 order_id，版本字段是 updated_at（同秒冲突再加 batch_id / CDC LSN），用 ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY updated_at DESC, batch_id DESC) 取最新，再 MERGE 回写。排序键必须确定且单调，否则同一批数据每次取到不同行，幂等就不成立。

Q3 「先删分区再插入」就幂等了吗？还缺什么？

▸ 答案：不够。还缺三样：① 事务/原子提交——删完插入失败，分区就空了；② 确定性——窗口、聚合、LIMIT 没有确定排序，重跑结果可能不同；③ 完整分区覆盖——只删了 dt=今天，但迟到数据落在 dt=前天，那部分仍会重复或不更新。生产上用 INSERT OVERWRITE 或「事务内 delete + insert」，并让分区范围由影响面计算得出。

Q4 一笔退款在 10-06 才到，订单日期是 10-01。要重算哪些分区？为什么不能只重算 10-06？

▸ 答案：要重算 10-01 起所有可能受影响的汇总分区（本例 10-01 的 DWS/ADS），以及任何累计指标（月累计、用户终身价值）。只重算 10-06 会让 10-01 的 GMV 永远停在退款前的值；而 10-06 本身可能根本没有这笔订单。重算范围由「事件日期 + 指标的回溯窗口」决定，通常按 7/30/90 天配置。

Q5 全量、增量、拉链，什么时候分别用？

▸ 答案：全量覆盖适合小维表、逻辑频繁变、或无法可靠识别变更的表；增量合并适合事实表（订单、支付、日志），前提是有稳定主键与更新时间；拉链适合「会变且要按历史时点回溯」的维度（类目、等级、归属）。三者可以组合：事实表增量、维度表拉链、汇总表按分区全量重算。

Q6 分层后报表反而变慢，问题通常出在哪？

▸ 答案：通常不是「层太多」，而是：① DWS/ADS 仍是视图，每次查询都重算；② 分区裁剪失效（查询没带 dt，或函数包住分区列）；③ 小文件过多，扫描计划开销大于计算；④ 宽表缺排序/聚簇，扫描量没降。解法是物化 + 分区裁剪 + 文件合并 + 列存排序，而不是把层删掉退回 ODS。相关原理见 44.16 OLAP 与列式存储。

Q7 用本节的数字说明：为什么「报表直接查 ODS」会虚高 93%？

▸ 答案：ODS 有 16 行，其中 8 行是重复投递，o2 还有一条旧金额 200 和新金额 210；报表按 status='paid' 直接求和，把两批都算进去，得到 1970 元。ADS 在 DWD 去重（9 单）后按口径算，毛 GMV 1020 元。差额 950 元来自重复行、旧版本和未收敛的状态。93.1% 不是极端案例，而是「重复投递 + 状态变更」在无分层时的常态。

关联