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

这篇具有很好参考价值的文章主要介绍了数据工程中的单元测试完全指南（含SQL代码）（下）。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

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

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

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

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 对象的值。请参考下表以获取此数据的表示方式。

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

在这种情况下，我们期望订单 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镜像、将代码部署到分段环境、运行端到端测试等内容。