dbt+SQLServer构建数据仓库(5)：数据测试

本文是系列第 5 篇,讲述怎么声明、dbt 怎么把 YAML 翻译成测试 SQL、测试失败怎么排查、自定义测试怎么加。

一、引言

数据测试是 dbt 的核心价值之一。

传统数仓方案靠"肉眼检查 + 抽样核对"保证数据质量,问题往往要等业务方报错才暴露。dbt 把测试变成了可声明、可自动运行、可阻断 CI 的代码:你在 YAML 里写一句 unique,dbt 自动生成 SQL、自动跑、返回行就报红。

本篇讲述如何进行数据测试,执行 dbt test 后全部 PASS。本文就以这些真实声明为案例,讲清楚测试体系。

二、测试类型:generic vs singular

dbt 的数据测试分两类:

generic test(通用测试):在 schema.yml 里声明,本质是"模板 + 参数"。dbt 内置四种模板(unique / not_null / relationships / accepted_values),也支持在 macros/ 里自定义新模板。声明时只需指定列名和测试名,dbt 自动套模板生成 SQL。

singular test(单例测试):在 tests/ 目录下写 .sql 文件,整段 SQL 完全手写。dbt 执行它,返回任意行即判定失败。

两者区别一句话:generic 是"填参数",singular 是"写全文"。

三、四种内置 generic test 详解

下面以项目里的真实声明为案例,逐种讲四种内置测试,并贴出 dbt 实际生成的测试 SQL(取自 target/compiled/ 目录)。

1. unique(唯一性)

检查某列的值是否唯一,常用于主键。

- name: customer_id
  description: "客户唯一标识."
  tests:
    - unique
    - not_null

dbt 生成的测试 SQL:

select
    customer_id as unique_field,
    count(*) as n_records

from "TEST"."dbt_dev_staging"."stg_customers"
where customer_id is not null
group by customer_id
having count(*) > 1

逻辑:按 customer_id 分组,筛出出现次数 > 1 的值。返回行即失败(说明有重复主键)。

2. not_null(非空)

检查某列是否存在 NULL。声明见 stg_customers 的 first_name:

- name: first_name
  tests:
    - not_null

dbt 生成的测试 SQL:

select first_name
from "TEST"."dbt_dev_staging"."stg_customers"
where first_name is null

逻辑:直接筛出 first_name is null 的行。返回行即失败。这是最简单的测试模板。

3. relationships(引用完整性)

检查外键值是否在引用表中存在,是跨模型的测试。

- name: customer_id
  description: "下单客户 ID, 关联 stg_customers.customer_id."
  tests:
    - not_null
    - relationships:
        to: ref('stg_customers')
        field: customer_id

dbt 生成的测试 SQL:

with child as (
    select customer_id as from_field
    from "TEST"."dbt_dev_staging"."stg_orders"
    where customer_id is not null
),

parent as (
    select customer_id as to_field
    from "TEST"."dbt_dev_staging"."stg_customers"
)

select
    from_field

from child
left join parent
    on child.from_field = parent.to_field

where parent.to_field is null

逻辑:子表 left join 父表,筛出父表匹配不到(parent.to_field is null)的行。返回行即失败(说明有断链的外键)。

注意:relationships 里的 ref('stg_customers') 会被 dbt 解析成真实表名。这意味着测试本身也是 DAG 节点,它依赖两个模型,两个模型都必须先构建完成才能跑这条测试。

4. accepted_values(枚举值)

检查列值是否都在允许列表内。:

- name: status
  description: "订单状态: placed / shipped / completed / returned / return_pending."
  tests:
    - accepted_values:
        values: ['placed', 'shipped', 'completed', 'returned', 'return_pending']

dbt 生成的测试 SQL(以 stg_payments 的 status 为例):

with all_values as (

    select
        status as value_field,
        count(*) as n_records

    from "TEST"."dbt_dev_staging"."stg_payments"
    group by status

)

select *
from all_values
where value_field not in (
    'completed','pending','refund'
)

逻辑:先按 status 分组聚合,再筛出不在允许列表里的值。返回行即失败(说明出现了非法枚举值,常见于源数据拼写错误)。

四、schema.yml 完整解读

这里示例测试声明分布在两个文件:

staging 层

完整内容:

version: 2

models:
  - name: stg_customers
    description: "客户 staging 视图, 1:1 投影 raw_customers."
    columns:
      - name: customer_id
        description: "客户唯一标识."
        tests:
          - unique
          - not_null
      - name: first_name
        tests:
          - not_null
      - name: last_name
        tests:
          - not_null

  - name: stg_orders
    description: "订单 staging 视图, 1:1 投影 raw_orders."
    columns:
      - name: order_id
        description: "订单唯一标识."
        tests:
          - unique
          - not_null
      - name: customer_id
        description: "下单客户 ID, 关联 stg_customers.customer_id."
        tests:
          - not_null
          - relationships:
              to: ref('stg_customers')
              field: customer_id
      - name: order_date
        tests:
          - not_null
      - name: status
        description: "订单状态: placed / shipped / completed / returned / return_pending."
        tests:
          - accepted_values:
              values: ['placed', 'shipped', 'completed', 'returned', 'return_pending']

  - name: stg_payments
    description: "支付 staging 视图, 1:1 投影 raw_payments."
    columns:
      - name: payment_id
        description: "支付唯一标识."
        tests:
          - unique
          - not_null
      - name: order_id
        description: "关联订单 ID."
        tests:
          - not_null
          - relationships:
              to: ref('stg_orders')
              field: order_id
      - name: payment_method
        tests:
          - accepted_values:
              values: ['credit_card', 'bank_transfer', 'coupon']
      - name: status
        tests:
          - accepted_values:
              values: ['completed', 'pending', 'refund']

marts 层

[models/marts/schema.yml] 完整内容:

version: 2

models:
  - name: dim_customers
    description: "客户维度表: 包含客户基础信息 + 订单聚合指标 (首末单日期、订单数、LTV)."
    columns:
      - name: customer_id
        description: "客户唯一标识, 主键."
        tests:
          - unique
          - not_null
      - name: first_name
        tests:
          - not_null
      - name: last_name
        tests:
          - not_null
      - name: first_order_date
        description: "客户首单日期, 无订单客户为 NULL."
      - name: most_recent_order_date
        description: "客户最近一次下单日期."
      - name: number_of_orders
        description: "客户累计订单数."
      - name: lifetime_value
        description: "客户生命周期价值 (已完成支付金额合计)."

  - name: fct_orders
    description: "订单事实表: 每行一个订单, 含订单状态与已完成支付金额."
    columns:
      - name: order_id
        description: "订单唯一标识, 主键."
        tests:
          - unique
          - not_null
      - name: customer_id
        description: "下单客户 ID, 关联 dim_customers.customer_id."
        tests:
          - not_null
          - relationships:
              to: ref('dim_customers')
              field: customer_id
      - name: order_date
        tests:
          - not_null
      - name: status
        tests:
          - accepted_values:
              values: ['placed', 'shipped', 'completed', 'returned', 'return_pending']
      - name: amount
        description: "订单已完成支付金额, 无已完成支付则为 0."

测试分布统计

逐 model 逐 column 统计,全项目 26 条测试分布如下:

模型	unique	not_null	relationships	accepted_values	合计
stg_customers	1	3	0	0	4
stg_orders	1	3	1	1	6
stg_payments	1	2	1	2	6
dim_customers	1	3	0	0	4
fct_orders	1	3	1	1	6
合计	5	14	3	4	26

测试的层次设计

观察上表,测试是分层复制的:

• staging 层测源数据质量:每个模型的主键 unique + not_null,状态/支付方式做 accepted_values,外键做 relationships。staging 是 raw 数据的 1:1 投影,这里测的是"源数据本身干不干净"。
• marts 层测业务一致性:dim_customers 和 fct_orders 重复了主键、非空、外键、枚举值测试。这里测的是"经过 JOIN/聚合后,业务表依然满足约束"。

两层都测看起来"重复",但目的不同:staging 测输入,marts 测输出。如果某条测试在 staging PASS 但在 marts FAIL,说明问题出在转换逻辑,而不是源数据。

五、dbt test 的执行流程

执行 dbt test 时,dbt 内部流程:

1. 解析 schema.yml:读取所有 models 下的 tests 声明。
2. 生成测试 SQL:为每条 generic test 套用模板,把 ref() 解析成真实表名,生成完整 SQL。
3. 构建测试 DAG:测试也是节点。relationships 测试依赖两个模型,必须等两个模型都构建完才能执行。
4. 按依赖顺序执行:逐条跑测试 SQL。
5. 判定结果:返回行数 > 0 即 FAIL,返回 0 行即 PASS。

输出格式大致如下:

✓ Pass  34  not_null_stg_customers_customer_id (stg_customers.customer_id) [0.05s]
✓ Pass  35  unique_stg_customers_customer_id (stg_customers.customer_id) [0.04s]
...
26 of 26 tests passed

关键点:测试是 DAG 的一部分。dbt test 会自动先构建被依赖的模型,再跑测试。所以 relationships 测试里写 ref('stg_customers'),dbt 会确保 stg_customers 先存在于库里。

六、测试失败怎么排查

测试 FAIL 时,dbt 默认只告诉你"哪条测试失败、返回多少行",看不到具体是哪些数据出问题。排查工具:

1. dbt test --store-failures

加 --store-failures 参数,dbt 会把失败的行存到 dbt_dev_staging schema 的表里,表名就是测试名。然后可以直接查:

dbt test --store-failures

-- 在 SQL Server 里查失败行
select * from "TEST"."dbt_dev_staging"."unique_stg_customers_customer_id"

这是最直接的排查手段——你能看到到底是哪个 customer_id 重复了。

2. dbt test --select 只测一个模型

排查时不想等全量 26 条跑完,可以只测一个模型:

dbt test --select stg_customers

也可以只测某一列的测试:

dbt test --select stg_customers.customer_id

3. 看编译后的测试 SQL

所有测试 SQL 都会编译到 target/compiled/ 目录,路径形如:

target/compiled/dbt_sqlserver_dw/models/staging/schema.yml/
  ├── unique_stg_customers_customer_id.sql
  ├── not_null_stg_customers_first_name.sql
  └── relationships_stg_payments_order_id__order_id__ref_stg_orders_.sql

直接打开这些 .sql 文件,能看到 dbt 生成的完整 SQL(本文第三节贴的就是这些),可以拷到 SSMS 里手动跑,定位问题。

4. 常见失败原因

• 源数据有重复:主键 unique 失败,通常是 raw 表里就有重复行。
• 源数据有 NULL:not_null 失败,通常是上游字段缺失。
• 外键断裂:relationships 失败,通常是子表引用了父表不存在的 ID(如订单引用了已删除的客户)。
• 枚举值拼写错误:accepted_values 失败,如 status 出现了 'shipped ' (多了空格)或 'SHIP'(大小写)。

排查流程图

┌──────────────┐
│  dbt test FAIL │
└──────┬───────┘
       │
       ▼
┌─────────────────────────────┐
│ dbt test --store-failures   │
│ --select <失败模型>          │
└──────┬──────────────────────┘
       │
       ▼
┌─────────────────────────────┐
│ 查 dbt_dev_staging 里的     │
│ 失败行表,看具体哪些数据出错 │
└──────┬──────────────────────┘
       │
       ▼
┌─────────────────────────────┐
│ 定位根因:                   │
│ - 源数据问题 → 修源/修 seed  │
│ - 转换逻辑问题 → 修模型 SQL  │
│ - 测试本身过严 → 调测试声明  │
└──────┬──────────────────────┘
       │
       ▼
┌─────────────────────────────┐
│ 重新 dbt test 验证          │
└─────────────────────────────┘

七、测试进阶

本项目只用了内置 generic test,但 dbt 的测试能力不止于此。

1. 自定义 generic test

在 macros/ 目录写一个 test_ 开头的 macro,即可在 schema.yml 里像内置测试一样引用。例如定义一个"金额必须为正"的测试:

-- macros/test_positive_value.sql
{% test positive_value(model, column_name) %}
    select *
    from {{ model }}
    where {{ column_name }} < 0
{% endtest %}

在 schema.yml 里引用:

- name: amount
  tests:
    - positive_value

dbt 会把 {{ model }} 替换成真实表名,{{ column_name }} 替换成 amount,生成测试 SQL。

2. singular test

在 tests/ 目录写任意 .sql 文件,例如 tests/assert_order_amount_positive.sql:

-- tests/assert_order_amount_positive.sql
select *
from {{ ref('fct_orders') }}
where amount < 0

执行 dbt test 时,dbt 会跑这段 SQL,返回行即失败。singular test 适合复杂断言,比如"所有已完成订单的支付金额合计应等于 fct_orders.amount"这种跨表核对。

3. severity:warn vs error

每条测试可配 severity:

• error(默认):失败即报错,阻断 CI。
• warn:失败只告警,不阻断。

- name: status
  tests:
    - accepted_values:
        values: ['placed', 'shipped', 'completed', 'returned', 'return_pending']
      severity: warn

适用场景:某些列存在已知脏数据,短期内无法修,但不想阻断流水线,先用 warn 监控。

4. 测试也是代码

测试声明应该:

• 入 git:和模型 SQL 一起版本管理,变更可 review。
• 在 CI 里跑:每次 PR 触发 dbt test,测试不过则 CI 失败,把数据问题挡在合并之前。

这是 dbt 测试体系相对传统方案的最大优势——数据质量保障从"人工抽检"升级为"自动化门禁"。

八、小结

数据测试是 dbt 把"数据质量"工程化的核心机制。本项目 26 条测试的要点:

1. 测试分两类:generic(模板+参数,在 schema.yml 声明)和 singular(完整 SQL,在 tests/ 写)。本项目 26 条全是 generic。
2. 四种内置模板:unique(查重复)、not_null(查空值)、relationships(查外键断链)、accepted_values(查枚举值)。覆盖了主键、非空、引用完整性、值域四类核心约束。
3. 测试是 DAG 节点:relationships 测试依赖两个模型,dbt 自动按依赖顺序执行。
4. 分层测试:staging 测源数据质量,marts 测业务一致性,两层都测以区分"源问题"和"转换问题"。
5. 排查三件套:--store-failures 存失败行、--select 缩小范围、target/compiled/ 看编译 SQL。
6. 进阶能力:自定义 generic test macro、singular test、severity 控制阻断行为,测试入 git + CI 跑。

写到这里,你已能为项目声明一套完整的数据测试并排查失败。但有个问题一直被"用着没深究":测试里的 ref('stg_customers') 是怎么被解析成真实表名的?整个 DAG 又是怎么自动构建的?下一篇回答这个。