如何在Jinja/宏条件分支中按需执行dbt插入模型

1. 编写检查数据存在性的宏 #

先创建一个宏来查询目标表中是否已存在指定日期的数据，放在macros/check_date_exists.sql文件中：

{% macro check_date_exists(target_table, date_column, input_date) %}
    {% set check_query %}
        SELECT COUNT(1) FROM {{ target_table }} WHERE {{ date_column }} = '{{ input_date }}'
    {% endset %}
    
    {% set check_result = run_query(check_query) %}
    {% if check_result.rows[0][0] > 0 %}
        {{ return(True) }}
    {% else %}
        {{ return(False) }}
    {% endif %}
{% endmacro %}

这个宏会执行计数查询，返回目标表中是否存在指定日期的记录。

2. 在插入模型中加入条件判断 #

在你的插入模型文件（比如models/insert_into_a.sql）中，调用上面的宏，只有当目标表无对应日期数据时才执行插入：

{% set input_date = var('input_date') %} -- 从外部传入的日期变量

{% set has_existing_data = check_date_exists(ref('a'), 'your_date_column', input_date) %}

{% if not has_existing_data %}
    INSERT INTO {{ ref('a') }} (col1, col2, your_date_column)
    SELECT 
        source_col1,
        source_col2,
        '{{ input_date }}' AS your_date_column
    FROM {{ ref('your_source_model') }}
    -- 这里可以添加其他过滤或转换逻辑
{% else %}
    -- 如果已有数据，执行空查询避免dbt报错
    SELECT 'Skip insertion: Date {{ input_date }} already exists in table a' AS execution_message
{% endif %}

关键说明 #

• ref('a')在这里可以正常使用，它会正确解析目标表的完整标识符（数据库、模式、表名），宏通过这个引用执行检查查询。
• run_query是dbt内置宏，用于执行SQL并返回结果，需确保你使用的dbt-snowflake适配器版本支持该功能。
• 传入日期变量的方式：执行dbt命令时通过--vars参数传入，比如：

  dbt run --select insert_into_a --vars '{"input_date": "2024-05-20"}'
  

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