鲁班 - 系统设计大全

60.5 数据质量

一句话

数据质量不是「多加几个告警」，而是把「可接受的数据」写成可执行契约，并在数据进入下一层前阻断；判据是错的数有没有流到决策者手里，不是告警响没响。

六个维度，六类规则

维度 回答的问题 可执行规则 典型漏检
完整性 该有的有没有 非空率、行数、分区是否存在 行数正常但某个渠道全空
唯一性 有没有重复 主键唯一、业务键去重率 重复行金额相同，DISTINCT 后看不出来
及时性 到得晚不晚 新鲜度（当前时间 − 最新分区时间）、产出时间 数据到了但内容是昨天的
有效性 格式/范围对不对 枚举、范围、正则、类型 金额为负、状态是拼写错误
一致性 跨表/跨层对不对 订单 vs 支付 vs 库存、DWD vs ADS 对账 两边都错但错得一样（同源 bug）
准确性 是否反映真实世界 与权威源抽样比对、业务规则校验 系统性地多算 10%，规则本身错

🔑 六维度里最难的是准确性：前五个都能用数据自身算出来，准确性必须有一个「外部真值」。所以生产上常用「对账 + 抽样 + 业务规则」三件套，而不是指望一条 SQL 查出来。一致性的分级口径（哪些允许最终一致、哪些必须强一致）见 46.3 数据一致性分级。

规则类型：能抓什么、抓不到什么

规则 能抓 抓不到
非空 解析失败、字段缺失 空值被填了默认值
唯一 重复投递、Join 放大 两条不同 ID 的重复业务
枚举 状态拼写、脏值 新增合法枚举未同步
范围 负数金额、异常年龄 合理范围内的错值
正则 手机号/身份证格式 格式对但号码是假的
引用完整性 孤儿外键 维表本身错了
跨表对账 金额/行数不一致 同源同错
波动 骤增骤降 缓慢漂移（每周降 1%）
新鲜度 上游延迟 数据到了但分区写错
Schema 兼容 列删除/类型收窄 语义变化（同名不同义）

⚠️ 只监控行数是数据团队最常见的安全感幻觉：行数正常、金额算错的事故，比「没数据」更难发现，也更难解释。

阈值为什么会疲劳：五种基线

方法 原理 优点 致命问题
固定阈值 null_rate > 1% 就告警 简单、可解释 季节性/促销/低流量全误报；调松就漏报
同比 / 环比 和上周同一天、昨天比 抵消周内规律 需要至少 2 周历史；遇节假日失效
3σ 偏离均值 3 个标准差 统计上「客观」 均值和标准差被异常值污染，异常越大越检不出
中位数 + MAD 偏离中位数 4 个「稳健标准差」 50% 崩溃点，抗异常值 数据分布突变（如业务改版）会连续误报
EWMA / 泊松 指数加权 / 低流量计数模型 适合低量、趋势 参数难调，解释成本高

3σ 的「掩蔽效应」值得单独记住：一个巨大的历史尖峰会把标准差撑大，让之后的真实下跌看起来「正常」。本节代码会实测给你看：30 天里注入 5 个异常，只配行数 3σ 的流水线命中 0/5，而中位数 + MAD 命中 5/5、误报 0。

代码：30 天、5 类异常，三种检测器对比

#!/usr/bin/env python3
"""数据质量引擎：固定阈值 vs 3σ vs 中位数+MAD（无第三方依赖）。"""
import random
import statistics as st

random.seed(42)
BASE = 1000


def make_days():
    days = []
    for i in range(1, 31):
        rows = BASE + random.randint(-20, 20)
        gmv = rows * random.randint(195, 205)
        days.append({"i": i, "rows": rows,
                     "null_rate": round(random.uniform(0.0005, 0.0015), 4),
                     "dup_rate": round(random.uniform(0.0001, 0.0008), 4),
                     "fresh_min": random.randint(10, 20), "gmv": gmv})
    days[19].update(rows=3000, gmv=600000)          # D20：大促，合法尖峰
    days[24].update(rows=600)                        # D25：上游漏数
    days[25].update(null_rate=0.06)                  # D26：字段解析失败
    days[26].update(gmv=days[26]["rows"] * 520)      # D27：金额口径错
    days[27].update(dup_rate=0.04)                   # D28：重复投递
    days[28].update(fresh_min=180)                   # D29：上游延迟
    return days


def zscore_flags(vals, k=3.0):
    """朴素 3σ：均值和标准差都会被异常值污染。"""
    if len(vals) < 5:
        return False
    mu, sd = st.mean(vals), st.pstdev(vals)
    return sd > 0 and abs(vals[-1] - mu) / sd > k


def mad_flags(vals, k=4.0):
    """中位数 + MAD：50% 崩溃点，少量异常值不会带偏基线。"""
    if len(vals) < 5:
        return False
    med = st.median(vals)
    mad = st.median([abs(x - med) for x in vals])
    sigma = 1.4826 * mad
    if sigma == 0:
        return vals[-1] != med
    return abs(vals[-1] - med) / sigma > k


def naive(days):
    """只配了「行数 3σ」的流水线。"""
    res = {}
    for idx in range(7, len(days)):
        hist = [d["rows"] for d in days[idx - 7:idx]]
        res[days[idx]["i"]] = zscore_flags(hist + [days[idx]["rows"]])
    return res


def robust(days):
    """多规则：量、空值率、重复率、金额、新鲜度。"""
    res = {}
    for idx in range(7, len(days)):
        d = days[idx]
        hist = days[idx - 7:idx]
        flags = []
        if mad_flags([x["rows"] for x in hist] + [d["rows"]]):
            flags.append(f"行数 {d['rows']}（基线 {int(st.median([x['rows'] for x in hist]))}）")
        if d["null_rate"] > max(0.02, 3 * st.median([x["null_rate"] for x in hist])):
            flags.append(f"空值率 {d['null_rate']:.2%}")
        if d["dup_rate"] > max(0.01, 3 * st.median([x["dup_rate"] for x in hist])):
            flags.append(f"重复率 {d['dup_rate']:.2%}")
        if mad_flags([x["gmv"] for x in hist] + [d["gmv"]]):
            flags.append(f"GMV {d['gmv']}（基线中位数 {int(st.median([x['gmv'] for x in hist]))}）")
        if d["fresh_min"] > 60:
            flags.append(f"新鲜度 {d['fresh_min']} 分钟")
        res[d["i"]] = flags
    return res


def main():
    days = make_days()
    n, r = naive(days), robust(days)
    print(f"{'日':>3} {'行数':>6} {'空值率':>7} {'重复率':>7} {'GMV':>8} {'延迟':>5}  "
          f"{'3σ':<4} 鲁棒检测")
    for d in days[20:]:
        print(f"{d['i']:>3} {d['rows']:>6} {d['null_rate']:>7.2%} {d['dup_rate']:>7.2%} "
              f"{d['gmv']:>8} {d['fresh_min']:>4}m  {'⚠' if n[d['i']] else '·':<4} "
              f"{'；'.join(r[d['i']]) or '·'}")

    injected = {25, 26, 27, 28, 29}
    calm = {21, 22, 23, 24, 30}
    print(f"\n注入异常 {len(injected)} 个；3σ 命中 "
          f"{sorted(i for i in injected if n[i])}（{sum(n[i] for i in injected)}/{len(injected)}），"
          f"误报 {sorted(i for i in calm if n[i])}")
    print(f"              鲁棒命中 "
          f"{sorted(i for i in injected if r[i])}（{sum(bool(r[i]) for i in injected)}/{len(injected)}），"
          f"误报 {sorted(i for i in calm if r[i])}")

    print("\n门禁判定：")
    for i in sorted(injected):
        p0 = any(k in " ".join(r[i]) for k in ("行数", "空值率", "重复率"))
        print(f"  D{i}: {'P0 阻断' if p0 else 'P1 告警'} -> {'；'.join(r[i])}")


main()

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

  日     行数     空值率     重复率      GMV    延迟  3σ   鲁棒检测
 21   1017   0.14%   0.04%   204417   13m  ·    ·
 22    988   0.10%   0.06%   200564   11m  ·    ·
 23    989   0.07%   0.06%   202745   19m  ·    ·
 24    984   0.09%   0.08%   197784   18m  ·    ·
 25    600   0.14%   0.01%   202188   11m  ·    行数 600（基线 993）
 26   1014   6.00%   0.03%   201786   14m  ·    空值率 6.00%
 27   1007   0.10%   0.08%   523640   14m  ·    GMV 523640（基线中位数 202188）
 28   1012   0.10%   4.00%   199364   20m  ·    重复率 4.00%
 29    999   0.10%   0.02%   204795  180m  ·    新鲜度 180 分钟
 30    990   0.15%   0.07%   200970   10m  ·    ·

注入异常 5 个；3σ 命中 []（0/5），误报 []
              鲁棒命中 [25, 26, 27, 28, 29]（5/5），误报 []

门禁判定：
  D25: P0 阻断 -> 行数 600（基线 993）
  D26: P0 阻断 -> 空值率 6.00%
  D27: P1 告警 -> GMV 523640（基线中位数 202188）
  D28: P0 阻断 -> 重复率 4.00%
  D29: P1 告警 -> 新鲜度 180 分钟

读数：

现象 数字 原因
3σ 全漏 0/5 D20 的 3000 行把 7 天窗口的标准差撑大，D25 的 600 行落在 3σ 内
MAD 全中 5/5 中位数与 MAD 不受单个尖峰影响，基线仍是 ~1000
3σ 误报 0 它的阈值太宽，宁可漏报也不误报
MAD 误报 0 平静日 D21–24、D30 都落在 4 个稳健标准差内
门禁分级 3 个 P0 / 2 个 P1 行数、空值率、重复率直接阻断；金额与延迟先告警

🔑 这个对比不是「MAD 比 3σ 好」，而是基线必须匹配数据分布：有尖峰、有季节性的业务数据，均值/标准差不是可靠的基线；而 MAD 也有失效场景（业务改版导致分布整体平移，会连续误报），所以要配「白名单 + 人工确认 + 规则分级」。

数据质量也有 SLI / SLO

SLI 定义 目标示例 失败时用户感知
新鲜度 现在 − 最新分区产出时间 P95 ≤ 30 min 看到昨天的数
完整度 实际行数 / 上游应有行数 ≥ 99.5% 报表少一截
Schema 兼容 不兼容变更数 / 总变更数 0 下游任务凌晨失败
质量通过率 通过的 P0 规则数 / 总 P0 规则数 100% 错的数进决策

🔑 数据 SLI 与接口 SLO 的差别：接口 SLI 看「请求成功没有」，数据 SLI 看「数对不对、新不新」。两者要分开定目标、分开报警，否则「任务成功」会掩盖「数据是错的」。

门禁：在哪一层阻断，阻断到什么程度

位置 检查 失败动作 理由
接入（ODS 前） schema、行数、新鲜度、重复率 P0 阻断入湖 脏数据进湖后，清理成本高一个数量级
DWD 发布前 主键唯一、非空、枚举、金额范围 P0 阻断发布 明细错了，所有下游都错
DWS / ADS 发布前 跨表对账、波动、口径断言 P0 阻断 / P1 告警 汇总错了会直接进报表
消费端 新鲜度、质量分、血缘 页面打标 / 拒绝查询 让用户知道「这数不可信」

门禁位置与 60.2 数仓分层 (ODS / DWD / DWS / ADS) 的层级一一对应：接入挡在 ODS 前，明细挡在 DWD 发布前，汇总挡在 ADS 前。阻断的代价要提前设计：

  • 隔离区：把不合格分区写到 quarantine，不覆盖主表，修好后重放；
  • 保留上一版：门禁失败时继续对外提供上一版数据，并在报表标注「数据延迟」；
  • on-call：P0 阻断必须有人接，否则第二天整条链路停摆；
  • 回滚开关：新规则上线先 warn，观察一周误报率再切 block。

⚠️ 「阻断」不是目的。没有隔离区与重放工具的门禁，最后一定会被降级成告警——因为业务不能停。告警分级、症状 vs 原因、runbook 的设计见 50.5 告警 (分级 / 症状 vs 原因 / runbook / 降噪)；质量 SLI 与错误预算见 50.7 SLO 与错误预算。

数据契约：把 schema 变更挡在发布前

变更 兼容性 处理
新增可空列 兼容 直接发布，通知下游
新增非空列（无默认） 不兼容 先加可空列 + 回填，再收紧
删除列 不兼容 先下线消费方，保留一个版本别名
改列名 不兼容 新增列 + 双写 + 迁移，最后删旧列
类型放宽（int → long） 兼容 直接发布
类型收窄（long → int） 不兼容 禁止；会截断
改语义（同名不同义） 最危险 必须换名 + 版本化，不能只改注释

契约的最小内容：schema + 语义 + SLA + owner + 兼容性策略。CI 里做 schema diff，不兼容变更直接拒绝合并；dbt 的 contract: enforced: true、Avro/Protobuf 的兼容性检查都是这个思路。兼容性分级与 Expand-Contract 的完整做法见 55.5 向后兼容。

对账：唯一能抓「系统性错误」的手段

对账 比什么 容差 差异分类
订单 vs 支付 订单数、金额 金额 0，行数 0（T+1 后） 未支付、重复支付、金额不符
订单 vs 库存 扣减次数、剩余库存 0 超卖、少卖、回补失败
DWD vs ADS 汇总金额、行数 0.01% 口径差异、过滤条件不同
源库 vs 数仓 行数、checksum 0（同分区） 漏采、重复、延迟

🔑 对账必须走独立链路：用同一张表、同一段 SQL 自我对账，只能发现「抄错」，发现不了「源头就错」。订单对支付要分别从两个系统取数，最好由不同团队实现。交易侧的复式记账与差异处理见 67.3 对账与差异处理（待补）。

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

指名：检查框架用 Great Expectations（替代 Soda、dbt tests 视栈而定）；转换层内嵌测试用 dbt tests；调度集成用 Airflow（GreatExpectationsOperator 或独立任务）。

可抄（Great Expectations 0.18.x，未验证：本机未安装 GE）：

# great_expectations/expectations/orders_suite.json（节选）
expectations:
  - expectation_type: expect_column_values_to_not_be_null
    kwargs: {column: order_id}
  - expectation_type: expect_column_values_to_be_unique
    kwargs: {column: order_id}
  - expectation_type: expect_column_values_to_be_in_set
    kwargs: {column: status, value_set: [paid, refunded, canceled, created]}
  - expectation_type: expect_table_row_count_to_be_between
    kwargs: {min_value: 800, max_value: 1200}

great_expectations checkpoint run orders_checkpoint
# 预期：Checkpoint succeeded；失败时进程退出码非 0，调度器据此阻断下游

dbt 侧的最小测试：

models:
  - name: dwd_trade_order_di
    columns:
      - name: order_id
        tests: [not_null, unique]
      - name: status
        tests:
          - accepted_values:
              values: ['paid', 'refunded', 'canceled', 'created']

可验：本节脚本输出「5/5 命中、0 误报」即检测器生效；生产上用「注入一条已知坏数据 → 门禁必须失败」做回归，而不是只看「今天没告警」。可退：新规则先 warn 一周，统计误报率；阻断开关按数据集配置，出问题时可单独降级为告警，并保留「上一版分区 + 质量标记」，让报表显式展示「数据不可信」。

反模式

反模式 后果
只监控行数 行数正常、金额算错，事故照样进报表
阈值拍脑袋（「空值率 1%」） 要么天天误报，要么永远不响
用 3σ 监控有尖峰/季节性的指标 尖峰撑大方差，真实异常被掩蔽（本节 0/5）
质量检查放在报表层 脏数据已经流到 DWD/DWS，清理成本翻倍
阻断但没有隔离区与重放 业务停摆，第二天门禁被关掉
用同一段 SQL 对账 同源同错，对账永远平
不做 schema 兼容检查 上游删列，下游凌晨全空
用「检查通过率」当质量分 只挑容易的规则，分数好看但关键规则缺失

自检

Q1 六个质量维度分别对应什么可执行规则？哪一维最难自动化？

▸ 答案：完整性→非空率/行数；唯一性→主键唯一/去重率；及时性→新鲜度/产出时间；有效性→枚举/范围/正则；一致性→跨表对账/跨层对账；准确性→与权威源抽样比对/业务规则。最难的是准确性，因为它需要「外部真值」，数据自身无法证明自己是对的，只能靠对账、抽样与业务验收。

Q2 为什么只配行数 3σ 的流水线在本节实测中 0/5 全漏？

▸ 答案：D20 的大促尖峰 3000 行落在 7 天窗口内，把均值拉高、标准差撑大（约 700），于是 D25 的 600 行偏离均值不到 3σ，被判为正常。异常值把「正常范围」撑宽了，这就是掩蔽效应（masking）。基线用均值/标准差时，越大的历史异常，越能让之后的真实异常「隐身」。

Q3 中位数 + MAD 为什么抗异常值？它在什么情况下会失效？

▸ 答案：中位数只依赖排序位置，MAD 是「到中位数距离的中位数」，两者都有 50% 崩溃点：少于一半的数据被污染时，基线基本不变。失效场景：① 业务整体平移（改版、口径变更）导致连续多天偏离，会连续误报；② 数据量很低且分布离散，MAD 可能为 0；③ 异常持续超过窗口一半，基线本身被带偏。

Q4 质量门禁应该放在哪一层？阻断还是告警？

▸ 答案：接入层挡 schema/行数/新鲜度/重复；DWD 发布前挡主键、非空、枚举、金额范围；DWS/ADS 发布前挡对账与波动；消费端做新鲜度与质量标记。阻断级别按「错了会不会直接改变决策」定：主键重复、金额为负、核心对账不平 → P0 阻断；金额偏移、延迟、长尾空值 → P1 告警并打标。阻断必须配隔离区、重放、on-call 与回滚开关，否则会被业务绕过。

Q5 哪些 schema 变更兼容，哪些不兼容？「改列名」为什么最危险？

▸ 答案：新增可空列、类型放宽兼容；删除列、改列名、类型收窄、新增无默认非空列不兼容。改列名危险在于：它既破坏旧查询（列不存在），又可能让新查询「看起来正常」（同名新列语义不同），而且历史文件/下游缓存里仍是旧名。正确做法是新增列 + 双写 + 迁移 + 下线旧列，而不是直接 rename。

Q6 为什么对账必须走独立链路？同源对账会漏什么？

▸ 答案：同源对账（同一张表、同一段 SQL、同一套过滤）只能验证「抄写一致性」，无法发现源头计算错误、漏采、口径错误。独立链路要求两边从不同系统/不同实现取数（订单系统 vs 支付系统），才能发现「同一笔业务在两个系统里不一致」。代价是要处理时间差、状态差与重复，需定义容差与差异分类。

Q7 质量告警太多，团队开始忽略，怎么降噪？

▸ 答案：① 分级：P0 阻断、P1 工单、P2 日报，不同级别走不同通道；② 基线化：用同比/MAD 替代固定阈值，按业务日历设白名单；③ 聚合：同一根因（上游延迟）只发一条，不按表刷屏；④ 责任到 owner，告警必须带「影响哪张报表/哪个指标」；⑤ 定期复盘误报，误报率高的规则下线或调参。降噪的目标不是「少发」，而是「发的每一条都可行动」。

关联