如何批量覆盖dbt测试命名规则？附测试结果存储需求

一、优雅修改测试命名（无需手动逐个设置name） #

针对source.yml中批量配置的源数据测试，可通过自定义宏覆盖dbt默认的测试命名逻辑，实现统一、结构化的命名规则：

1. 编写自定义测试命名宏 #

在dbt项目的macros目录下创建test_name.sql文件，基于测试类型、源表、字段动态生成命名：

{% macro test_name(test, node) %}
    -- 从测试内置名称提取核心类型（如not_null、unique）
    {% set test_type = test.name.split('_')[0] %}
    {% set source_name = node.source_name %}
    {% set table_name = node.name %}
    -- 兼容表级测试（无字段）的场景
    {% set column_name = node.columns[0].name if node.columns else 'table' %}
    
    -- 生成清晰命名：源名_测试类型_表名_字段名
    {{ source_name }}_{{ test_type }}_{{ table_name }}_{{ column_name }}
{% endmacro %}

dbt运行时会自动调用该宏生成测试名称，替代默认带哈希值的混乱命名，比如生成jaffle_shop_not_null_customer_customer_id这类直观名称。

2. 差异化命名规则扩展 #

如果需要针对特定源或表调整命名格式，可在宏中添加条件判断：

{% macro test_name(test, node) %}
    {% if node.source_name == 'jaffle_shop' %}
        -- 针对jaffle_shop源的简化命名
        js_{{ test.name.split('_')[0] }}_{{ node.name }}_{{ node.columns[0].name if node.columns else 'table' }}
    {% else %}
        -- 默认通用规则
        {{ node.source_name }}_{{ test.name.split('_')[0] }}_{{ node.name }}_{{ node.columns[0].name if node.columns else 'table' }}
    {% endif %}
{% endmacro %}

二、测试结果存储的更佳方案 #

1. 利用on-run-end钩子批量写入 #

在dbt_project.yml中配置运行结束钩子，将dbt自动生成的dbt_test_results临时视图数据存入自定义结果表：

on-run-end:
  - "INSERT INTO your_schema.dbt_data_quality_results (
      test_name,
      test_status,
      error_message,
      execution_time
    ) SELECT
      name,
      status,
      message,
      executed_at
    FROM dbt_test_results"

注意：需提前在PostgreSQL中创建目标表，字段需与dbt_test_results的输出匹配（可通过dbt test --store-failures命令查看临时视图结构）。

2. 增量加载的测试结果模型 #

创建专门的dbt模型实现测试结果的增量存储与维度拆分，方便后续分析：

{{ config(
    materialized='incremental',
    schema='your_schema'
) }}

SELECT
    name AS test_name,
    status AS test_status,
    message AS error_message,
    executed_at AS execution_time,
    -- 从自定义测试名拆分分析维度
    split_part(name, '_', 1) AS source_system,
    split_part(name, '_', 2) AS test_category,
    split_part(name, '_', 3) AS table_name,
    split_part(name, '_', 4) AS column_name
FROM dbt_test_results
{% if is_incremental() %}
    -- 仅加载新的测试结果，避免重复
    WHERE executed_at > (SELECT COALESCE(MAX(execution_time), '1970-01-01') FROM {{ this }})
{% endif %}

运行dbt run --select dbt_data_quality_results即可增量更新结果表，便于后续统计唯一性、有效性指标的通过率趋势。

3. 进阶：结合元数据丰富结果维度 #

如果需要更详细的测试元数据（如测试描述、依赖关系），可在钩子中利用dbt的graph变量获取信息：

{% set test_nodes = graph.nodes.values() | selectattr('resource_type', 'equalto', 'test') %}
{% for test in test_nodes %}
    INSERT INTO your_schema.dbt_test_metadata (
        test_name,
        source_name,
        table_name,
        test_description
    ) VALUES (
        '{{ test.name }}',
        '{{ test.depends_on.nodes[0].split('.')[1] }}',
        '{{ test.depends_on.nodes[0].split('.')[2] }}',
        '{{ test.description | replace("'", "''") }}'
    );
{% endfor %}

将这段逻辑加入on-run-end钩子，即可把测试的描述、依赖源等元数据存入表中，提升分析维度。

内容的提问来源于stack exchange，提问作者Kingloss404