如何用dbt测试数据表行数始终不低于历史值？

完全可以通过dbt自定义测试实现这个需求，核心思路是对比当前表的行数与历史加载周期的行数，确保当前行数不会减少。以下是具体实现步骤：

1. 建立历史行数统计模型 #

首先创建一个增量模型，用来记录每日目标表的行数，方便后续对比。比如创建models/daily_row_counts.sql：

{{ config(materialized='incremental') }}

select
    current_date() as load_date,
    (select count(*) from {{ ref('your_target_table') }}) as row_count
from dual

{% if is_incremental() %}
    -- 增量模式下只插入当天的统计数据
    where load_date > (select max(load_date) from {{ this }})
{% endif %}

这个模型会每天自动插入一条记录，包含当天日期和目标表的总行数。

2. 编写自定义行数校验测试 #

在dbt的tests/generic目录下创建自定义测试宏row_count_not_decrease.sql：

{% test row_count_not_decrease(model, historical_table) %}
    select
        current_row_count,
        previous_row_count
    from (
        select
            -- 获取当前目标表的行数
            (select count(*) from {{ model }}) as current_row_count,
            -- 获取最近一次历史统计的行数，若为空则设为0（兼容首次运行）
            (select coalesce(max(row_count), 0) from {{ historical_table }}) as previous_row_count
    ) counts
    -- 当当前行数小于历史行数时，测试失败
    where current_row_count < previous_row_count
{% endtest %}

3. 在目标表中启用测试 #

在目标表对应的schema.yml配置文件中，添加这个自定义测试：

models:
  - name: your_target_table
    tests:
      - row_count_not_decrease:
          historical_table: ref('daily_row_counts')

额外注意事项 #

• 确保daily_row_counts模型的运行顺序在目标表之后，可以在schema.yml中通过depends_on明确依赖关系，避免统计行数时目标表还未完成加载。
• 如果你的数据仓库不支持dual表（比如BigQuery），可以替换成select 1或者其他等价写法。

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