Airflow中从DAG文件直接调用dbt宏的问题求助

问题描述 #

我尝试在Airflow的DAG文件中直接调用dbt宏作为任务（而非从模型中触发调用），但当前代码未达到预期效果。使用的代码如下：

run_macro = DbtTaskGroup(
    group_id="dbt_run_macro",
    project_config=ProjectConfig(
        dbt_project_path="/usr/local/airflow/dags/dbt/project_folder"
    ),
    profile_config=profile_config,
    execution_config=execution_config,
    operator_args={"select": "run-operation"},
    render_config=RenderConfig(
        select=["path:macros"]
    ),
)

配置render_config后，代码无法渲染任何宏；但去掉render_config时，会选中所有模型和快照。macros文件夹包含7-8个宏，且在Airflow环境外使用dbt run-operation macro_name命令可正常运行。

需求：希望从DAG文件中成功执行全部或指定的dbt宏。

问题根源 #

1. RenderConfig的select参数用于筛选dbt的标准资源（模型、快照、种子等），宏不属于这类可渲染资源，因此path:macros无法被识别。
2. operator_args中传入的select参数不符合dbt run-operation的语法——该命令不需要select，而是直接指定宏名。

解决方案 #

方案1：执行单个指定宏 #

直接使用DbtRunOperationOperator（或在DbtTaskGroup中明确指定命令），传入目标宏名即可：

run_macro = DbtTaskGroup(
    group_id="dbt_run_macro",
    project_config=ProjectConfig(
        dbt_project_path="/usr/local/airflow/dags/dbt/project_folder"
    ),
    profile_config=profile_config,
    execution_config=execution_config,
    dbt_task="run-operation",  # 明确指定执行run-operation命令
    operator_args={"macro_name": "your_target_macro"}  # 替换为实际宏名
)

方案2：执行macros文件夹下所有宏 #

手动遍历宏文件，为每个宏创建独立的DbtRunOperationOperator并加入TaskGroup：

import os
from airflow.providers.dbt.cloud.operators.dbt import DbtRunOperationOperator
from airflow.utils.task_group import TaskGroup
from airflow.providers.dbt.cloud.config import ProjectConfig

dbt_project_path = "/usr/local/airflow/dags/dbt/project_folder"
macros_dir = os.path.join(dbt_project_path, "macros")

with TaskGroup(group_id="dbt_run_all_macros") as run_all_macros:
    # 遍历macros目录下的所有.sql宏文件
    for filename in os.listdir(macros_dir):
        if filename.endswith(".sql"):
            # 从文件名提取宏名（假设文件名与宏定义名一致）
            macro_name = os.path.splitext(filename)[0]
            DbtRunOperationOperator(
                task_id=f"run_macro_{macro_name}",
                project_config=ProjectConfig(dbt_project_path=dbt_project_path),
                profile_config=profile_config,
                execution_config=execution_config,
                macro_name=macro_name
            )

关键注意事项 #

• DbtTaskGroup默认针对模型类资源设计，run-operation属于特殊命令，直接使用DbtRunOperationOperator逻辑更清晰。
• 确保指定的macro_name与宏文件中定义的名称完全一致（dbt对宏名大小写敏感）。
• 若宏需要传入参数，可通过operator_args的args参数传递，示例：

  operator_args={"macro_name": "your_macro", "args": {"param1": "value1", "param2": 123}}
  

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