数据工程中的单元测试完全指南（含SQL代码）（下）

详细讲解数据工程中的单元测试。

微信搜索关注《Java学研大本营》

3 单元测试 SQL 数据模型

进行基于 SQL 的数据模型单元测试主要集中在测试 SQL 查询上。通过测试 SQL 查询，您可以验证在给定特定输入时，查询是否返回预期结果，以及它是否能正确处理边缘情况。这有助于确保您的数据模型在各种情况下都能正常运行，并提供正确的结果。

考虑一个计算平均订单价值的SQL查询：

SELECT AVG(order_value) as avg_order_value FROM orders;

针对这个查询的单元测试可以向

orders

表插入几行数据，并验证查询是否正确计算了平均值。虽然有一些工具可以促进SQL单元测试，但在数据工程领域中最受欢迎的工具之一是dbt（data build tool数据构建工具），它内置支持数据测试。

在这个示例中，假设你已经设置好了一个dbt项目，并且按照dbt的推荐项目结构进行操作。你会有一个计算平均订单值的模型（我们称之为

avg_order_value.sql

），我们可以为它添加一个测试。

首先，在

avg_order_value.sql

中定义模型：

-- 这是计算平均订单值的模型
SELECT AVG(order_value) as avg_order_value FROM {{ ref('orders') }}

对于测试，使用dbt的数据测试。这些只是dbt针对你的数据运行的SQL查询。如果查询返回任何行，测试将失败。如果查询不返回任何行，测试将通过。

在

tests

目录中创建一个名为

avg_order_value_test.sql

的新文件。在这个文件中，我们将编写一个查询，检查平均订单值是否在预期范围内：

-- 这是你的测试
-- 假设你已经手动计算了测试数据的平均值，为100

WITH avg_order_value_test AS (
  SELECT AVG(order_value) as avg_order_value FROM {{ ref('orders') }}
)

SELECT *
FROM avg_order_value_test
WHERE avg_order_value <> 100 -- 如果平均订单值不为100，测试将失败

记得将100替换为你的测试数据的预期平均订单值。

你可以使用命令

dbt test

来运行这个测试。

这是一个简单的例子，但你可以构建更复杂的测试场景。例如，你可以向

orders

表添加更多数据，并验证平均值是否正确更新，或者你可以添加一些边缘情况，比如订单价值为0或NULL，并确保它们被正确处理。

测试COUNT： 假设你有一个SQL查询，用于计算表中活跃用户的数量：

SELECT COUNT(*) as active_users FROM users WHERE status = 'active';

对于这个查询，一个单元测试会向

users

表中插入一些测试数据，确保其中一些用户的状态为“active”。然后执行查询，并验证返回的计数是否符合预期。

使用dbt，COUNT示例的基于YAML的测试如下所示：

version: 2

models:
  - name: users
    tests:
      - dbt_utils.expression_is_true:
          expression: "count(*) = 5"
          where: "status = 'active'"

在这里，我们断言活跃用户的数量恰好为5。根据你的测试数据设置进行相应的修改。

测试连接： 假设你有一个连接两个表（

orders

和

order_items

）的查询，计算每个订单的总成本：

SELECT o.order_id, SUM(i.item_price * i.quantity) as total_order_cost
FROM orders o 
JOIN order_items i ON o.order_id = i.order_id
GROUP BY o.order_id;

为了对这个查询进行单元测试，首先要使用测试数据填充

orders

和

order_items

表，确保某些订单有多个商品。然后测试将验证每个订单的

total_order_cost

是否正确计算，特别是那些有多个商品的订单。

使用dbt，JOIN示例的基于YAML的测试如下所示：

version: 2

models:
  - name: orders
    tests:
      - dbt_utils.expression_is_true:
          expression: "total_order_cost = 100.00"
          where: "order_id = 1"

在这里，我们断言

order_id

为1的订单总成本为100。根据您的测试数据设置进行相应调整。

在这个例子中，我们希望验证我们的连接操作和随后的聚合操作是否正常工作。为了创建一个更复杂的测试，考虑一个场景：我们有多个订单，每个订单中有多个商品，我们想验证多个订单的总成本计算是否正确。

假设我们的测试数据设置如下：

订单表：

[
  {
    "order_id": 1,
    "customer_id": 100
  },
  {
    "order_id": 2,
    "customer_id": 101
  },
  {
    "order_id": 3,
    "customer_id": 102
  }
]

订单项表：

[
  {
    "order_item_id": 1,
    "order_id": 1,
    "item_price": 10,
    "quantity": 5
  },
  {
    "order_item_id": 2,
    "order_id": 1,
    "item_price": 20,
    "quantity": 2
  },
  {
    "order_item_id": 3,
    "order_id": 2,
    "item_price": 30,
    "quantity": 3
  },
  {
    "order_item_id": 4,
    "order_id": 2,
    "item_price": 40,
    "quantity": 1
  },
  {
    "order_item_id": 5,
    "order_id": 3,
    "item_price": 50,
    "quantity": 1
  }
]

在 JSON 表示中，表中的每一行都变成了数组中的一个 JSON 对象。列名成为 JSON 对象的键，相应的单元格值成为 JSON 对象的值。请参考下表以获取此数据的表示方式。

在这种情况下，我们期望订单 1 的总成本为90（50 + 40），订单 2 的总成本为130（90 + 40），订单 3 的总成本为50（50*1）。

我们的 dbt 基于 YAML 的测试可以如下所示：

version: 2

models:
  - name: orders
    tests:
      - dbt_utils.expression_is_true:
          expression: "total_order_cost = 90.00"
          where: "order_id = 1"
      - dbt_utils.expression_is_true:
          expression: "total_order_cost = 130.00"
          where: "order_id = 2"
      - dbt_utils.expression_is_true:
          expression: "total_order_cost = 50.00"
          where: "order_id = 3"

在这里，我们添加了多个断言来检查多个订单的总订单成本的正确性。这使我们不仅可以验证连接操作的正确性，还可以验证我们能否正确计算多个订单和商品的总成本。

4 单元测试与持续集成

一旦您拥有一套单元测试，您可以将它们集成到持续集成（CI）流水线中。CI流水线会在代码库中的更改被推送时自动构建和测试您的代码。这样，如果某个更改在您的数据流水线或数据模型中引入了错误，CI流水线将在影响生产数据系统之前捕捉到它。

有几种CI工具可以帮助您进行设置，例如Jenkins、Travis CI或GitLab CI。设置CI流水线的具体细节将取决于工具，但总体思想是在每次推送新代码时运行单元测试。

看看如何使用流行的CI/CD工具GitHub Actions将Python单元测试集成到简单的CI流水线中。

假设您有一个使用pytest编写的一组单元测试的Python项目。您有一个名为

test_example.py

的文件，其中包含以下函数：

def test_addition():
    assert 1 + 1 == 2

通过创建一个GitHub Actions工作流程，您可以在每次将更改推送到GitHub存储库时自动运行这些测试。在您的存储库中创建一个名为

.github/workflows/ci.yml

的新文件，内容如下：

name: Python CI

on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

jobs:
  build:

    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v2

    - name: Set up Python
      uses: actions/setup-python@v2
      with:
        python-version: '3.8'

    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install pytest
        # 如果您的项目在requirements.txt中列出了其他依赖项，请使用以下命令安装它们：
        # pip install -r requirements.txt

    - name: Run tests
      run: |
        pytest

该工作流程在每次将更改推送到

main

分支或对

main

分支发起拉取请求时运行。它设置了一个全新的Python 3.8环境，安装了您项目的依赖项，并运行了pytest测试套件。如果您的任何测试失败，工作流程将失败，并会通知您失败的原因。

这只是一个简单的示例。真实的CI流水线会更加复杂，并包括构建Docker镜像、将代码部署到分段环境、运行端到端测试等内容。

5 总结

在数据可信度至关重要的领域中，虽然单元测试需要额外工作，但这些努力将带来长期的收益。对于数据流水线和数据模型来说，建立信心很重要。

数据工程可以从软件工程方法中获益，这是关键所在。采用单元测试和持续集成等实践，可以提升数据系统的可靠性，及时发现错误，避免错误传播到下游，并最终在组织中树立对数据的信任文化。

这些测试的目标不仅是验证在正常情况下的正确性，还包括确保您的流水线或查询能够优雅地处理边界情况和错误。通过进行单元测试，您可以构建强大而可靠的数据流水线和SQL数据模型，增强对数据系统的信心。

数据工程中的单元测试完全指南（含SQL代码）（下）

3 单元测试 SQL 数据模型

4 单元测试与持续集成

5 总结

推荐书单

《 MySQL从入门到精通（第3版）（软件开发视频大讲堂）》

精彩回顾

发表评论

“数据工程中的单元测试完全指南（含SQL代码）（下）”的评论:

关于作者

overfit同步小助手

相关阅读

文章导航