跳至主要内容

dbt 测试全指南:确保数据质量

了解在 dbt 中用于检查并确保数据质量的各类测试。
· 11分钟 阅读

使用 AI 探索

dbt 因其数据转换与测试功能在分析领域日益流行。本文将介绍 dbt 是什么,以及如何开展 dbt 测试来确保数据质量。若您刚开始使用 dbt,欢迎查看我们的dbt 入门课程以进一步学习。 

什么是 dbt?

正如我们在另一篇指南中所探讨的,dbt(data build tool) 是一款广受欢迎的开源工具,供数据团队将仓库中的原始数据转换为干净、经过测试、有完整文档且可直接用于分析的数据集。 

与ETL 工具不同,dbt 专注于数据管道中的 “T”(转换)环节,并运行在 BigQuery、Snowflake、Redshift、Databricks 等现代云数据仓库之上。

dbt 最强大的功能之一是其内置的数据测试框架,使数据从业者能够编写测试来验证关于数据的假设。该测试框架不仅有助于尽早发现问题,对于在不断增长、模块化的分析代码库中维护数据质量与信任也至关重要。

为何在 dbt 中进行测试?

测试是分析工程的基石。随着数据集规模增长、更多相关方依赖数据,未被发现的数据问题成本也将上升。dbt 的测试功能可帮助您:

  • 及早发现数据问题:在下游仪表板与报告受到影响之前。
  • 自动化质量检查:避免手动校验工作。
  • 促进良好的数据建模实践:推动模块化、可测试的转换。
  • 支持 CI/CD 实践:无缝集成至自动化工作流。
  • 落实数据契约:验证关键约束(如不得重复)能长期成立。

dbt 测试的类型

在 dbt 中,测试是基于 SQL 的断言,用于验证数据假设。dbt 中的测试主要分为两类:

1. 通用(内置)测试

这些测试由 dbt 预定义,并以声明式方式在YAML 文件中应用。通用测试覆盖数据建模与数据仓库中常见的约束。 

其特点包括:

  • 配置简便
  • 可在多列与多模型间复用
  • 对简单检查有高性能

四种内置的通用测试类型包括:

  • not_null:确保某列不包含任何 NULL 值。
  • unique:确保某列各行取值彼此唯一(常用于主键)。
  • accepted_values:将某列限制在一组预定义的允许值内。
  • relationships:验证表间引用完整性,确保外键能在另一张表中找到匹配值。

这些测试非常适合用于强制执行基础的数据完整性,尤其是围绕键、分类型字段与核心表关系。

通用测试的工作原理是自动在底层生成 SQL。 

例如,对 customer_id 列进行 not_null 测试会生成类似如下的 SQL:

SELECT *
FROM {{ ref('customers') }}
WHERE customer_id IS NULL

若该查询返回任何行,测试即判定失败。

2. 自定义(独立)测试

自定义或独立测试是用户定义的 SQL 查询,保存为 .sql 文件,位于 tests/ 目录。这些测试用于:

  • 通用测试无法表达的复杂校验
  • 业务规则的执行
  • 跨模型逻辑
  • 统计或条件阈值

独立测试提供最大灵活性。只要 SQL 查询仅返回违反预期的行,几乎可以以任何方式验证数据。

例如:

  • 确保产品的上线日期不晚于其退市日期
  • 检测异常高或异常低的值(离群点检测)
  • 验证汇总值与业务口径总量一致

由于具备高度灵活性,自定义测试非常适合分析团队在生产环境中执行细致入微的数据质量契约。

dbt 中包含的宏也可用于测试,例如 dbt-utils 包。

小结:

特性

通用测试

自定义(独立)测试

定义位置

schema.yml

tests/ 文件夹下的 .sql 文件

覆盖范围

常见约束(如空值、键)

任何可用 SQL 表达的逻辑

复杂度

简单

中等至复杂

可复用性

高

低(通常特定场景)

灵活性

有限

无限(任意 SQL 逻辑)

结合两种测试可以取长补短:通用测试带来一致性与覆盖面,自定义测试提供精确度。下面我们分别展开介绍。 

1. 通用(内置)测试

通用测试由 dbt 预定义,您只需将其添加到模型的 schema.yml 文件中,即可以声明式方式使用。 

这些测试通常用于验证唯一性、非空、引用完整性或是否属于定义集合等约束。

下面用一个简单教程来实践 dbt 中的通用测试。

如何声明通用测试

在模型对应的 schema.yml 文件中,于 columns: 区块下定义测试:

version: 2

models:
  - name: customers
    description: "Customer dimension table"
    columns:
      - name: customer_id
        description: "Unique customer identifier"
        tests:
          - not_null
          - unique
      - name: email
        tests:
          - not_null

示例:not_null 与 unique

yaml
columns:
  - name: customer_id
    tests:
      - not_null
      - unique

上述测试确保 customer_id 在每一行均存在且唯一。它常用于强制 customer_id 作为该表的主键这一假设。

示例:accepted_values

columns:
  - name: customer_type
    tests:
      - accepted_values:
          values: ['new', 'returning', 'vip']

该测试检查 customer_type 字段仅包含三种允许的字符串:new、returning 或 vip。这类测试常用于必须符合既定取值集合的分类型字段,如枚举或状态。

示例:relationships

columns:
  - name: customer_id
    tests:
      - relationships:
          to: ref('customers')
          field: id

该测试通过校验当前模型中的每个 customer_id 均可在 customers 表中作为 id 找到,来强制执行引用完整性约束。它在分析层模拟了 SQL 的外键约束。

2. 自定义(独立)测试

当内置测试无法满足您的用例时(例如需要验证复杂业务逻辑),您可以使用 SQL 编写自定义测试,即独立测试。

如何创建自定义测试

1. 在 dbt 项目的 tests/ 目录内创建一个 .sql 文件。

2. 编写一个仅返回测试失败行的 SQL 查询。

示例:orders 中不得有未来日期

在 tests 文件夹内创建名为 no_future_dates.sql 的文件。

SELECT *
FROM {{ ref('orders') }}
WHERE order_date > current_date

该测试检查 orders 表中是否存在 order_date 位于未来的记录。若返回任何行,测试即失败,并提醒您存在无效数据,可能由时区错误、ETL 缺陷或源系统录入不当导致。

示例:按地区去重邮箱

-- 文件:tests/duplicate_emails_per_region.sql

SELECT email, region, COUNT(*) as occurrences
FROM {{ ref('customers') }}
GROUP BY email, region
HAVING COUNT(*) > 1

该测试确保在给定地区内每个邮箱唯一。这可能反映了某项业务规则:同一人不能在同一区域重复注册。任何返回的行均表示数据质量违规。

如何在 dbt 中实施测试

本节将逐步演示如何在您的项目与部署流水线中实施、配置并运行 dbt 测试。

步骤 1:设置 dbt

在开始编写测试或模型之前,您需要安装 dbt 并初始化一个新项目。

操作步骤:

1. 创建项目文件夹:

在任意位置创建一个文件夹。

2. 进入项目文件夹:

cd dbt-test-project

3. 创建 Python 虚拟环境:

python3 -m venv dbt-env

随后,在环境创建完成后将其激活。

dbt-venv\Scripts\activate

4. 安装 dbt

pip dbt install

5. 创建 .dbt 文件夹

mkdir $home\.dbt

6. 初始化 dbt

dbt init

7. 创建 profiles.yml 文件

在 .dbt 文件夹中新建一个文件,内容如下:

austi:
  target: dev
  outputs:
    dev:
      type: sqlite
      threads: 1
      database: ./dbt_project.db
      schema: main
      schema_directory: ./schemas
      schemas_and_paths:
        main: ./dbt_project.db

您可以将“austi”替换为您 Windows 电脑的用户配置文件名称。

8. 创建 dbt_project.yml 文件

接下来,您需要在 .dbt 文件夹中新建另一个配置文件,内容如下。

name: dbt_test_project
version: '1.0'
profile: austi

seeds:
  dbt_test_project:
    +quote_columns: false

同样,将“austi”替换为您的用户配置文件名称。

9. 验证项目可用:

dbt debug

完成设置后,您就可以开始创建数据集并构建 dbt 模型了。

步骤 2:创建数据集

如果您无法访问数据仓库,可以使用 CSV 文件与 dbt 的 seed 功能在本地模拟 dbt 测试。

操作步骤:

1. 创建 CSV 文件:将其置于 dbt 项目的 /seeds 新文件夹内。

命名方式如下:

seeds/customers.csv

customer_id,name,email,customer_type
1,Alice Smith,[email protected],new
2,Bob Jones,[email protected],returning
3,Carol Lee,[email protected],vip
4,David Wu,[email protected],new

2. 在同一目录内再创建一个文件:

使用以下命名方式:

seeds/orders.csv

order_id,customer_id,order_date,order_total,order_status
1001,1,2023-12-01,150.00,shipped
1002,2,2023-12-03,200.00,delivered
1003,1,2023-12-05,175.00,cancelled
1004,3,2024-01-01,225.00,pending

3. 创建用于识别 seeds 的配置文件:

接下来,您需要创建名为 dbt_project.yml 的配置文件。

将以下内容粘贴至该配置文件中。 

name: dbt_test_project
version: '1.0'
profile: austi

seeds:
  dbt_test_project:
    +quote_columns: false

将 profile 字段更改为与您 Windows 电脑的用户配置文件名称一致。

4. 加载 seed 数据:

dbt seed

该命令会根据 CSV 文件创建 main.customers 与 main.orders 表。它们是替代数据库所需的种子数据。

loading seed data

如上图所示,已找到并加载了两个 seed。

步骤 3:导入至 dbt

现在我们来创建分层(staging)模型,以便转换并测试原始数据。

操作步骤:

1. 创建分层模型:

在项目文件夹中新建 models 文件夹。在该文件夹内创建名为 stg_customers.sql 的文件。

将以下代码粘贴至文件中:

SELECT
  *
FROM {{ ref('customers') }}

在另一个名为 stg_orders.sql 的文件中,粘贴以下代码:

SELECT
  *
FROM {{ ref('orders') }}

2. 定义 schema 测试:

在以下位置创建新文件:

models/schema.yml

将以下内容粘贴至该文件

version: 2

models:
  - name: stg_customers
    description: "Customer staging table"
    columns:
      - name: customer_id
        description: "Unique identifier for each customer"
        tests:
          - not_null:
              tags: [critical]
          - unique:
              tags: [critical]

      - name: email
        description: "Email address of the customer"
        tests:
          - not_null

  - name: stg_orders
    description: "Orders staging table"
    columns:
      - name: order_id
        description: "Unique identifier for each order"
        tests:
          - not_null:
              tags: [critical]
          - unique:
              tags: [critical]

      - name: customer_id
        description: "Foreign key to stg_customers"
        tests:
          - not_null:
              tags: [critical]

3. 运行 dbt 构建分层模型:

dbt run --select stg_customers stg_orders

这些分层模型现在可作为使用本地 seed 数据开展 dbt 测试的基础。

示例输出如下:

building staging models

步骤 4:创建您的第一个测试

数据已完成分层,接下来让我们使用内置的通用测试框架实现您的首个 dbt 测试。

操作步骤:

1. 打开定义了分层模型的 schema.yml 文件。

2. 在模型的 columns: 区块下,定义如 not_null 或 unique 等测试。

示例:

models:

  - name: stg_customers

    columns:

      - name: customer_id

        tests:

          - not_null

          - unique
  1. 使用 CLI 运行测试:
dbt test --select stg_customers

预期结果如下:

testing for nulls and uniques

步骤 5:开发自定义与独立测试

当您的需求超出内置测试范围时,可使用 SQL 创建自定义测试。

操作步骤:

1. 在 tests/ 文件夹中新建文件,例如:tests/no_future_orders.sql

2. 添加返回失败行的 SQL 逻辑:

SELECT *
FROM {{ ref('stg_orders') }}
WHERE order_date > current_date

3. 运行测试:

dbt test --select no_future_orders

若存在任何未来日期的 order_date,测试将失败。

预期结果如下:

testing for no future orders

步骤 6:在 CI/CD 流水线中运行测试

为确保测试执行的一致性,请将 dbt 测试集成至您的开发流水线。

操作步骤:

1. 使用标签来设定优先级或隔离测试:

您可以按以下格式修改任意模型的 schema 来包含这些标签:

columns:
  - name: customer_id
    tests:
      - not_null:
          tags: [critical]

2. 在本地选择性运行测试:

dbt test --select tag:critical

若测试正确运行,您应能看到如下输出:

testing for critical tags

3. 配置 CI 工作流以自动运行 dbt 测试。您可以将其连接到 GitHub Actions 等 CI 平台。 

dbt 测试最佳实践

高效实施 dbt 测试不仅在于编写测试,更在于将测试融入团队的开发周期与文化。以下是一些可供参考的最佳实践:

1. 从主键开始

始终对维表与事实表的主键列应用 not_null 与 unique 测试。这两类约束是下游查询中可靠关联与去重逻辑的基础。

2. 使用 relationships 验证外键

外键对于维护引用完整性至关重要。使用 relationships 测试来模拟外键约束,尤其是在未原生强制外键的数据仓库中。

3. 对分类型列使用 accepted_values

通过在 status、region 或 product_type 等列中强制特定取值,来保持维度一致性。这可降低拼写错误或未处理枚举扩展的风险。

4. 为业务逻辑编写自定义测试

业务逻辑并不总能与通用规则一一对应。以下是应编写自定义测试的示例:

  • 销售额不得为负
  • 订单日期不得在未来
  • 同一用户不能拥有多份在订的订阅

5. 清晰组织测试

当测试数量增多时,容易变得杂乱。请参考以下做法保持条理:

  • 将自定义测试放入 tests/ 文件夹,并使用具描述性的命名。
  • 避免过度复杂的 SQL——让逻辑聚焦且易读。
  • 尽可能按模型或领域对测试分组。

6. 快速失败并融入 CI/CD

在 CI 流水线中运行 dbt test。这可确保在数据质量约束不通过时,代码无法合并。将测试集成到 CI/CD 中可强化数据管道的责任意识与信任。

7. 避免过度测试

过多或不必要的测试(尤其是针对海量数据)会拖慢部署。聚焦以下测试:

  • 对业务至关重要
  • 可能因上游问题而失败
  • 在失败时有助于排障

除非属于契约性 SLA 的一部分,否则避免对计算字段进行测试。

进阶示例:离群点检测

作为进阶测试示例,我们可以进行一次离群点检测测试。

在 tests 文件夹内创建一个 SQL 文件,包含以下代码:

SELECT *
FROM {{ ref('orders') }}
WHERE order_total > 100000

这是一种基础的离群点检测。如果您的业务通常订单金额低于 $10,000,则可将超过 $100,000 的订单标记为人工复核对象。尽管这未必是严格的数据质量违规,但对于欺诈检测或运营监控仍很有价值。

性能考量

随着 dbt 项目规模扩大、测试覆盖增加,性能愈发重要。本节将介绍保持测试高效、控制计算成本的技术与策略。

测试优化技巧

测试时可尝试以下优化技巧:

  • 增量测试:借助 dbt 的增量模型结构,将测试聚焦于新添或修改的数据,减少对全量数据集的不必要扫描。
  • 查询并行化:在 dbt_project.yml 中配置 threads: 参数,利用 dbt 内置的并行执行以并发运行多个测试。
  • 选择性执行测试:在开发过程中使用 --select 与 --exclude 等标志,仅运行相关测试。例如:
dbt test --select customers
dbt test --exclude tag:slow

成本管理策略

实施成本管理策略有助于在与查询复杂度与频率相关的计费环境中,保持 dbt 测试的高性能与高性价比。

可采取以下策略:

  • 基于标签的测试执行:为资源密集型或可选测试打标签,使其在常规运行中被排除,除非明确需要。
  • 结果缓存:利用数据仓库特性(如 BigQuery 或 Snowflake 的结果缓存)尽可能避免重复计算。组织测试使其能受益于先前查询的缓存结果。
  • 计划性批量测试:将较重的测试套件安排在业务低峰时段或生产窗口之外,以尽量降低对仓库工作负载的影响。

结语

dbt 测试是确保高数据质量、自动化校验并在转换流程早期发现问题的有力手段。结合通用与自定义测试,可构建稳健、易维护的数据工作流。

在我们的dbt 入门课程或dbt 教程中了解更多 dbt 相关内容。

dbt 测试常见问题

dbt 中的 unique 测试是什么?

在 dbt(data build tool)中,unique 测试用于确保指定列中的所有值彼此不同,即不存在重复项。

dbt 自带的四种通用测试是什么?

dbt 自带四种测试:not_null、unique、accepted_values 和 relationships。

有哪些 dbt 测试类型?

dbt 测试类型包括通用内置测试与自定义独立测试。

Austin Chia

作者

Austin Chia

我是 Austin,一名博主和科技作者,曾在医疗领域担任数据科学家与数据分析师,拥有多年的从业经验。我以生物学背景开启技术之路,如今通过我的科技博客帮助他人完成同样的转型。我对技术的热情促使我为数十家 SaaS 公司撰写内容,分享经验、启发他人。

主题