如何将DBT脚本的日期结果返回至Airflow并实现循环调度？

针对你的需求，下面提供两种落地方案——一种结合Airflow实现循环调度，另一种纯用DBT包装器完成日期驱动的模型执行，均适配Snowflake环境：

一、Airflow + DBT：获取日期后循环执行目标脚本 #

DBT本身没有直接返回结果到外部系统的机制，但可以通过生成中间表或自定义宏输出两种方式让Airflow拿到日期列表，再循环调用目标DBT脚本。

1. 方案1：用DBT模型生成待处理日期表 #

先写一个DBT模型，从Snowflake中筛选出需要处理的日期（比如未加载的日期）：

-- models/staging/stg_pending_dates.sql
select distinct date_column::date as run_date
from raw_data.daily_table
where processed_flag = false
order by run_date

运行这个模型后，结果会存入Snowflake的对应表（比如analytics.staging.stg_pending_dates）。

然后在Airflow中读取该表的日期：

from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook

def get_pending_dates():
    hook = SnowflakeHook(snowflake_conn_id='your_snowflake_conn')
    sql = "select run_date from analytics.staging.stg_pending_dates"
    records = hook.get_records(sql)
    # 转成字符串格式，方便后续传递给DBT
    return [row[0].strftime('%Y-%m-%d') for row in records]

最后用Airflow的动态任务映射循环执行DBT：

from airflow.operators.bash import BashOperator
from airflow.decorators import dag, task
from datetime import datetime

@dag(start_date=datetime(2024,1,1), schedule_interval='@daily', catchup=False)
def dbt_date_driven_load():
    @task
    def fetch_dates():
        return get_pending_dates()

    target_dates = fetch_dates()

    # 动态生成每个日期对应的DBT任务
    run_dbt_task = BashOperator.partial(
        task_id='run_dbt_for_date',
        bash_command='cd /path/to/your/dbt/project && dbt run --models your_target_model --vars \'{"target_date": "{{ params.date }}"}\''
    ).expand(params=[{"date": d} for d in target_dates])

dbt_date_driven_load()

目标DBT模型里要接收target_date变量，比如：

-- models/marts/fact_daily_data.sql
select *
from raw_data.daily_table
where date_column = '{{ var("target_date") }}'

2. 方案2：用DBT run-operation直接输出日期 #

写一个自定义宏，查询Snowflake并打印日期：

-- macros/get_pending_dates.sql
{% macro get_pending_dates() %}
    {% set query %}
        select distinct date_column::varchar as run_date
        from raw_data.daily_table
        where processed_flag = false
    {% endset %}
    {% set results = run_query(query) %}
    {% if results %}
        {% for row in results %}
            {{ row[0] }}
        {% endfor %}
    {% endif %}
{% endmacro %}

在Airflow中执行dbt run-operation并捕获输出：

@task
def fetch_dates_via_dbt():
    # 执行DBT操作并捕获标准输出
    import subprocess
    result = subprocess.check_output(
        ['cd /path/to/dbt/project && dbt run-operation get_pending_dates'],
        shell=True,
        text=True
    )
    # 分割输出成日期列表
    return [d.strip() for d in result.split('\n') if d.strip()]

后续同样用动态任务映射执行目标DBT脚本即可。

二、纯DBT包装器：不依赖Airflow的日期调度 #

如果不需要Airflow，直接用DBT的宏和内置函数实现日期驱动的模型循环执行：

1. 编写日期获取宏 #

-- macros/get_target_dates.sql
{% macro get_target_dates() %}
    {% set query %}
        select distinct date_column::varchar as run_date
        from raw_data.daily_table
        where processed_flag = false
    {% endset %}
    {% set results = run_query(query) %}
    {% set dates = [] %}
    {% for row in results %}
        {% do dates.append(row[0]) %}
    {% endfor %}
    {{ return(dates) }}
{% endmacro %}

2. 编写包装器宏，循环执行目标模型 #

-- macros/run_model_for_dates.sql
{% macro run_model_for_dates(model_name) %}
    {% set target_dates = get_target_dates() %}
    {% if target_dates %}
        {% for date in target_dates %}
            {% do log(f"Processing {model_name} for date: {date}", info=True) %}
            -- 构造DBT运行命令并执行
            {% set run_cmd = "dbt run --models " ~ model_name ~ " --vars '{\"target_date\": \"" ~ date ~ "\"}'" %}
            {% do run_command(run_cmd) %}
        {% endfor %}
    {% else %}
        {% do log("No pending dates to process.", info=True) %}
    {% endif %}
{% endmacro %}

3. 执行包装器宏 #

在命令行运行：

dbt run-operation run_model_for_dates --args '{model_name: "your_target_model"}'

关键注意事项 #

• 日期格式统一：Snowflake的日期类型要和DBT、Airflow中传递的字符串格式一致（推荐YYYY-MM-DD），避免类型转换错误。
• 模型幂等性：目标模型要保证重复运行同一日期时不会重复生成数据，比如用Snowflake的MERGE语句替代INSERT。
• 权限控制：执行DBT的账户需要有Snowflake的查询权限，Airflow的Snowflake连接需要配置正确的角色和仓库。

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