Data quality validator

何时使用

• 为数据管道/数仓建立可重复、可报警的质量校验：用 Great Expectations 写期望套件、用 dbt test 写表/列级测试、用数据契约约束上下游 schema 与 SLA。
• 把校验接入 CI/CD 或调度（Airflow/dbt Cloud），让脏数据在进入下游前被拦截，失败时报警（Slack 等）。
• 监控质量指标：完整性、唯一性、有效性、准确性、一致性、新鲜度。
• 触发词：数据质量、data quality、Great Expectations/GE、dbt test、数据契约/data contract、校验规则、完整性、唯一性、新鲜度。

不该用：

• 单机离线 CSV/Excel 表的清洗（去重、缺失填充、类型规整）→ 用 csv-data-cleaner。
• 纯查询/联表/聚合建模本身 → 用 sql-query-builder（本技能只校验其产物质量）。
• 表外数据（API/文件原貌）的 schema 推断与解析。

步骤 / 指令

1. 定维度：先按六维（完整性/唯一性/有效性/准确性/一致性/新鲜度）圈定要测的关键列，不要全表全列乱测。
2. 分层测（数据测试金字塔，自下而上）：
   • schema 测试（结构）：列集合、列类型、行数区间。
   • 单列单元测试：非空、唯一、取值集合、数值/日期区间。
   • 跨表集成测试：外键关系、跨列约束、A>B 之类一致性。
3. 选工具：仓内 dbt 优先用 dbt test（声明在 _models.yml，配合 dbt_utils）；需要丰富断言/统计期望/数据文档用 Great Expectations；跨团队边界用数据契约（datacontract.com + SodaCL）固化 schema 与 SLA。
4. 落新鲜度：必测 created_at 之类的时间列在近 N 天/小时内，陈旧数据等于坏数据。
5. 编排+报警：用 GE checkpoint 或 dbt 在 CI/调度中批量跑，任一关键表失败即让管道失败并发通知。
6. 演进：契约/schema 版本化，发现新问题就增量补测，阈值用动态基线而非硬编码。

示例

GE 安装与最小套件（核心命令保留自源）：

pip install great_expectations
great_expectations init
great_expectations datasource new

import great_expectations as gx
context = gx.get_context()
suite = context.add_expectation_suite("orders_suite")
suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeUnique(column="order_id"))
# 取值集合 / 区间 / 新鲜度 / 行数 等同理：
# ExpectColumnValuesToBeInSet(column="status", value_set=[...])
# ExpectColumnValuesToBeBetween(column="amount", min_value=0, strict_min=True)
# ExpectColumnMaxToBeBetween(column="created_at", min_value="now-1d", max_value="now")
result = context.run_checkpoint(checkpoint_name="orders_checkpoint")
if not result.success:
    raise ValueError("Data quality checks failed!")

dbt 声明式测试（models/marts/.../_models.yml）：

models:
  - name: fct_orders
    tests:
      - dbt_utils.recency: {datepart: day, field: created_at, interval: 1}  # 新鲜度
    columns:
      - name: order_id
        tests: [unique, not_null]
      - name: customer_id
        tests:
          - not_null
          - relationships: {to: ref('dim_customers'), field: customer_id}   # 外键
      - name: order_status
        tests:
          - accepted_values: {values: [pending, processing, shipped, delivered, cancelled]}
      - name: total_amount
        tests:
          - dbt_utils.expression_is_true: {expression: ">= 0"}

自定义通用 dbt 测试（行数区间）：

{% test row_count_in_range(model, min_count, max_count) %}
with row_count as (select count(*) as cnt from {{ model }})
select cnt from row_count where cnt < {{ min_count }} or cnt > {{ max_count }}
{% endtest %}

数据契约骨架（datacontract.com + SodaCL 质量块）：

apiVersion: datacontract.com/v1.0.0
kind: DataContract
metadata: {name: orders, version: 1.0.0, owner: data-platform-team}
schema:
  properties:
    order_id:  {type: string, format: uuid, required: true, unique: true}
    customer_id: {type: string, required: true, pii: true}
    status:    {type: string, enum: [pending, processing, shipped, delivered, cancelled]}
quality:
  type: SodaCL
  specification:
    checks for orders:
      - missing_count(order_id) = 0
      - duplicate_count(order_id) = 0
      - freshness(created_at) < 24h
sla: {availability: 99.9%, freshness: 1 hour}

注意事项

• 聚焦关键列，别测一切；过度测试拖慢管道又难维护。
• 早测、增量测：变换前先校验源数据；遇到问题再补测，并为每个测试写清晰描述。
• 别忽略 warning，它常是 failure 前兆；别跳过新鲜度校验。
• 阈值用动态基线，勿硬编码；行数/均值类统计期望要随业务波动调整。
• 测关系而非孤立列：外键、跨表一致性比单列断言更能抓真实坏账。
• 失败必报警（接监控/Slack），并让 CI/调度真正失败而非仅记录。
• 契约/schema 版本化，schema 变更可追踪；PII 列在契约中标注且不得下泄到下游 mart。

互见

• requires：无。
• related：csv-data-cleaner（单机离线表的清洗，与入库后的质量校验互补）；sql-query-builder（被校验对象多由 SQL 建模产出）。
• combines_with：无。

---

本条采编自 wshobson/agents（MIT）。