未记录事实核验 · 状态：维护中

记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。

dbt 测试与数据质量 - 数据质量即代码

---

为什么要自动化测试

一个典型的场景：

周一早上，业务方发现上周的销售额报表数据有问题，订单总量少了 20%。 数据团队花了 3 小时定位：是因为某个 ETL 作业失败了，有 20% 的数据没有同步过来。 问题发生在周五晚上，整个周末没人发现。

如果有自动化测试：

# 每次 dbt 运行后自动检查
models:
  - name: fct_orders
    tests:
      - dbt_utils.recency:        # 检查最新数据不超过 24 小时前
          datepart: hour
          field: created_at
          interval: 24

在已配置定时执行和告警的条件下，测试失败可触发通知；发现延迟取决于执行频率及通知链路。该 recency 示例只检查最新记录时间，不能单独识别部分订单漏同步。

---

Schema 测试（内置四种）

dbt 内置了四种基础测试，直接在 YAML 文件中声明：

1. unique：唯一性测试

# models/marts/core/_core_models.yml

version: 2

models:
  - name: fct_orders
    description: "订单事实表，每行代表一个订单"
    columns:
      - name: order_id
        description: "订单唯一标识"
        tests:
          - unique         # 每个 order_id 必须唯一
          - not_null       # order_id 不能为空

这会生成并执行：

-- dbt 自动生成的 unique 测试 SQL
SELECT order_id, COUNT(*) AS cnt
FROM analytics.fct_orders
WHERE order_id IS NOT NULL
GROUP BY order_id
HAVING COUNT(*) > 1
-- 如果结果不为空，测试失败

2. not_null：非空测试

- name: customer_id
  tests:
    - not_null    # 每个订单必须有 customer_id

3. accepted_values：枚举值测试

- name: financial_status
  description: "订单支付状态"
  tests:
    - accepted_values:
        values:
          - 'pending'
          - 'paid'
          - 'refunded'
          - 'voided'
        # quote: false   # 如果值是数字，设置 quote: false

4. relationships：外键完整性测试

- name: customer_id
  tests:
    - relationships:
        to: ref('dim_customers')    # 引用的维度表
        field: customer_id          # 维度表中的字段
        # 验证：fct_orders 中的每个 customer_id 都能在 dim_customers 中找到

---

完整的 schema.yml 示例

# models/marts/core/_core_models.yml

version: 2

models:
  - name: fct_orders
    description: |
      订单事实表。每行代表一个已完成的客户订单。
      包含订单的所有度量值和外键引用。
      粒度：订单级别（order_id 唯一）

    tests:
      # 表级别测试（在 tests 列表下，不在 columns 里）
      - dbt_utils.expression_is_true:
          expression: "total_price_cents >= 0"    # 金额不能为负

    columns:
      - name: order_id
        description: "订单唯一标识，来源于 Shopify 的 orders.id"
        tests:
          - unique
          - not_null

      - name: customer_id
        description: "客户 ID，外键"
        tests:
          - not_null
          - relationships:
              to: ref('dim_customers')
              field: customer_id

      - name: financial_status
        description: "支付状态：pending/paid/refunded/voided"
        tests:
          - not_null
          - accepted_values:
              values: ['pending', 'paid', 'refunded', 'voided']

      - name: total_price_cents
        description: "订单总金额（分），包含税费和运费"
        tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: ">= 0"    # 金额不能为负

      - name: created_at
        description: "订单创建时间（UTC）"
        tests:
          - not_null

  - name: dim_customers
    description: "客户维度表，每行代表一个客户"
    columns:
      - name: customer_id
        tests:
          - unique
          - not_null
      - name: email
        tests:
          - unique       # 邮箱也应该唯一
          - not_null

---

测试的严重级别配置

不是所有测试失败都应该终止流程，可以配置严重级别：

columns:
  - name: email
    tests:
      - unique:
          severity: warn    # 邮箱重复只警告，不报错
      - not_null:
          severity: error   # 邮箱为空则报错（默认）

或者在测试中设置允许的失败阈值：

- name: order_id
  tests:
    - unique:
        error_if: ">100"    # 默认计数方式下，重复键分组数超过 100 才报错
        warn_if: ">10"      # 默认计数方式下，重复键分组数超过 10 就警告

---

自定义测试

当内置四种测试不够用时，可以自定义：

方式一：在 tests/ 目录创建 SQL 测试

-- tests/assert_orders_not_too_old.sql
-- 测试：没有超过 10 年的订单（数据异常检测）

SELECT order_id
FROM {{ ref('fct_orders') }}
WHERE created_at < CURRENT_DATE - INTERVAL '10 years'

-- 如果这个查询返回任何行，测试失败

方式二：创建可复用的测试宏

-- macros/test_is_positive.sql
-- 自定义测试：验证某列的值是正数

{% test is_positive(model, column_name) %}

SELECT {{ column_name }}
FROM {{ model }}
WHERE {{ column_name }} <= 0

{% endtest %}

在 YAML 中使用：

- name: total_price_cents
  tests:
    - is_positive    # 使用自定义的 is_positive 测试

方式三：使用 dbt-expectations 包（推荐）

dbt-expectations 是仿照 Great Expectations 设计的 dbt 测试扩展包，提供了几十种丰富的测试：

# packages.yml（先安装）
packages:
  - package: calogica/dbt_expectations
    version: [">=0.10.0", "<0.11.0"]

dbt deps    # 安装包

# schema.yml 中使用 dbt-expectations

models:
  - name: fct_orders
    tests:
      # 行数测试：订单表应该有足够的数据
      - dbt_expectations.expect_table_row_count_to_be_between:
          min_value: 1000
          max_value: 10000000

      # 数据新鲜度：最新记录不超过 24 小时前
      - dbt_expectations.expect_row_values_to_have_recent_data:
          timestamp_column: created_at
          datepart: hour
          interval: 24
          row_condition: "financial_status = 'paid'"

    columns:
      - name: total_price_cents
        tests:
          # 值域范围测试
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0
              max_value: 100000000    # 最大 100 万元（以分为单位）

          # 分布测试：95% 的值应该 > 0
          - dbt_expectations.expect_column_quantile_values_to_be_between:
              quantile: 0.05
              min_value: 1

      - name: email
        tests:
          # 格式测试：邮箱格式验证
          - dbt_expectations.expect_column_values_to_match_regex:
              regex: "^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$"

---

测试策略：什么时候测什么

分层测试策略

层次	测试重点	示例测试
Sources	数据新鲜度、行数合理	freshness 检查、not_null
Staging	主键唯一、字段格式	unique、not_null、regex
Intermediate	关联完整性、无意外空值	relationships、not_null
Marts	业务规则、数据量级	accepted_values、row_count

最小化但有效的测试清单

不需要测试每一列，聚焦在关键保障上：

# 每个模型至少要有这些测试：

# 1. 主键测试（必须）
- unique
- not_null

# 2. 外键完整性（重要关联的）
- relationships

# 3. 枚举值（状态、类型字段）
- accepted_values

# 4. 金额/数量的合理性
- dbt_expectations.expect_column_values_to_be_between

# 5. 数据新鲜度（最终层）
- dbt_expectations.expect_row_values_to_have_recent_data

---

运行测试

# 运行所有测试
dbt test

# 只运行某个模型的测试
dbt test --select fct_orders

# 只运行 unique 类型的测试
dbt test --select test_name:unique

# 运行 staging 层所有模型的测试
dbt test --select staging

# 同时运行模型和测试（推荐的 CI 用法）
dbt build    # 按依赖图执行所选范围内的 seed、snapshot、model 和 test 等资源

---

测试结果处理

# 测试失败时的输出
dbt test --select fct_orders
# ...
# FAIL 1 unique_fct_orders_order_id ........................ [FAIL in 1.23s]
#
# Failure in test unique_fct_orders_order_id (models/marts/core/_core_models.yml)
#   Got 15 results, configured to fail if != 0
#   compiled code at target/compiled/.../unique_fct_orders_order_id.sql

查看具体失败的数据：

# 找到 compiled 的测试 SQL
cat target/compiled/my_project/models/marts/core/unique_fct_orders_order_id.sql

# 在数据库中执行这个 SQL，看哪些 order_id 是重复的

---

数据质量即代码的理念

"如果你不能测试它，你就无法信任它。"

在软件工程里，代码没有测试是不被接受的。数据工程应该遵循同样的标准：

传统数据质量工作：

• 人工定期抽查报表
• 业务发现问题后回溯
• 依赖个人经验和责任心

dbt 测试驱动的数据质量：

• 测试和数据模型一起提交到 Git
• 每次 dbt build 自动运行（CI/CD 中）
• 配合调度和告警系统，在测试失败后通知相关人员，减少依赖业务发现问题的情况

与 数据质量评估方法 中的框架结合，dbt 测试可以覆盖：

• 完整性：not_null、row_count
• 唯一性：unique
• 一致性：relationships（外键完整性）
• 有效性：accepted_values、regex 格式检查
• 时效性：freshness check、recency

---

总结

测试类型	使用场景	实现方式
unique	主键、唯一键	内置
not_null	关键字段非空	内置
accepted_values	枚举状态字段	内置
relationships	外键完整性	内置
自定义 SQL	复杂业务规则	tests/ 目录
复用测试宏	跨模型通用规则	macros/ 目录
dbt-expectations	丰富的统计和格式测试	第三方包

---

下一步

测试为已覆盖的数据质量规则提供可重复检查，但不保证整体数据可信；接下来让数据可理解：

07-文档与数据血缘 - 用 dbt 自动生成数据文档和血缘图，让数据治理有据可查

---

最后更新：2026-03-18

#dbt #数据质量 #测试 #schema测试 #数据治理

---

---

掌握检查

• [ ] 能在 schema.yml 中为某个模型的主键字段配置 unique 和 not_null 测试，并用 dbt test --select 模型名 运行通过
• [ ] 能写一个自定义 SQL 测试文件（放在 tests/ 目录），验证某个业务规则（如"订单金额必须 > 0"）
• [ ] 知道 dbt build 和分别运行 dbt run + dbt test 的区别（build 按依赖图调度模型及测试，error 级别测试失败可使相关下游节点被跳过）
• [ ] 能说出 severity: warn 的使用场景：测试失败时不阻断流程，只输出警告
• [ ] 理解"最小有效测试集"的原则：每个模型至少要有主键的 unique + not_null 测试，而不是测试每一列

---

实践练习

基础题：为你的 fct_orders.sql 在 schema.yml 中添加三类测试：主键字段的 unique + not_null，状态字段的 accepted_values（只允许特定枚举值），以及一个外键完整性 relationships 测试（订单的 customer_id 必须存在于 customers 表）。运行 dbt test 验证全部通过。

进阶题：安装 dbt-expectations 包（packages.yml 中添加依赖后运行 dbt deps），为你的某个事实表添加两个统计测试：总金额字段的均值不能低于某个阈值（expect_column_mean_to_be_between），以及每日数据量相比昨日不能下降超过 30%（编写自定义 SQL 测试，按明确的日期口径计算当日与昨日行数并比较；需约定昨日无数据及当日尚未结束时的处理方式。expect_table_row_count_to_be_between 本身只检查行数上下界，不能直接完成昨日基线比较）。