如何将Airflow的执行日期传递给DBT Cloud脚本？

我正在通过Airflow执行DBT Cloud转换任务，当前任务调用代码如下：

trigger_job = DbtCloudRunJobOperator(
    task_id="trigger_dbt_cloud_job",
    dbt_cloud_conn_id=DBT_CLOUD_CONN_ID,
    job_id=JOB_ID,
    check_interval=600,
    timeout=3600,
)

希望了解如何将Airflow的执行日期传递至DBT脚本，恳请提供示例代码。

要把Airflow的执行日期传递给DBT脚本，你可以通过DbtCloudRunJobOperator的additional_run_config参数传递自定义变量，具体分两步操作：

1. 修改Airflow任务代码，传递执行日期 #

借助Airflow的模板变量，把执行日期打包进additional_run_config传给DBT Cloud：

from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator

DBT_CLOUD_CONN_ID = "your_dbt_cloud_conn_id"
JOB_ID = 12345  # 替换成你的DBT Cloud Job ID

trigger_job = DbtCloudRunJobOperator(
    task_id="trigger_dbt_cloud_job",
    dbt_cloud_conn_id=DBT_CLOUD_CONN_ID,
    job_id=JOB_ID,
    check_interval=600,
    timeout=3600,
    # 把Airflow执行日期作为DBT变量传递
    additional_run_config={
        "variables": {
            "airflow_execution_date": "{{ ds }}",  # 格式是YYYY-MM-DD
            # 也可以用无横杠的格式：{{ ds_nodash }} → YYYYMMDD
            # 或者完整的带时分秒的时间：{{ execution_date.isoformat() }}
        }
    },
    # 开启模板渲染，确保Airflow能解析变量
    template_fields=["additional_run_config"],
)

2. 在DBT脚本里接收并使用变量 #

在你的DBT模型SQL文件中，用var()函数获取这个变量，直接用于逻辑处理：

-- 示例：models/your_model.sql
SELECT
    id,
    user_name,
    order_time
FROM
    raw.orders
WHERE
    -- 筛选执行日期当天的数据
    order_time >= '{{ var("airflow_execution_date") }}'::DATE
    AND order_time < '{{ var("airflow_execution_date") }}'::DATE + INTERVAL '1 day'

注意事项 #

• 如果你需要传递其他格式的时间，直接替换模板变量即可，比如{{ execution_date }}会返回完整的datetime对象字符串。
• DBT Cloud默认允许接收自定义变量，不需要额外配置Job权限。

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