Airflow中如何向BigQueryOperator传递含变量的宏参数？

核心问题 #

你遇到的问题是：直接将带模板语法的字符串传入params，Airflow只会将其作为普通字符串渲染，不会二次解析模板；而嵌套{{}}的写法违反Jinja2语法，导致报错。

方法1：在SQL模板中直接使用参数运算（推荐） #

不需要在Python代码里拼接宏字符串，直接把days作为参数传入params，Jinja2支持在宏函数中直接使用params变量，无需嵌套模板标记。

步骤：

1. 循环初始化Operator时，传递days参数：

for days in range(0, 3):  # 加载最近3天数据
    bq_to_gcs = BigQueryOperator(
        task_id=f"task_id_{days}",
        sql=path.join("queries", "query.sql"),
        params={
            "schema": BQ_SOURCE_DATA_SET_ID,
            "days": days,  # 直接传入days数值
        }
    )

2. 在query.sql中直接调用宏并传入参数：

{{ macros.ds_format(macros.ds_add(ds, - params.days), '%Y-%m-%d', '%Y/%m/%d') }}

Jinja2会正确解析宏逻辑：先计算ds_add(ds, -params.days)得到目标日期，再格式化为指定路径格式。

方法2：提前计算日期（适用于已知执行日期的场景） #

如果可以在Python代码中提前获取执行日期（比如从DAG上下文或固定日期），可以直接计算好格式化后的日期字符串，再传入params：

from datetime import datetime, timedelta

for days in range(0, 3):
    # 获取执行日期，这里可根据实际场景替换为DAG的execution_date
    execution_date = datetime.now().date()
    target_date = execution_date - timedelta(days=days)
    date_path = target_date.strftime("%Y/%m/%d")
    
    bq_to_gcs = BigQueryOperator(
        task_id=f"task_id_{days}",
        sql=path.join("queries", "query.sql"),
        params={
            "schema": BQ_SOURCE_DATA_SET_ID,
            "date_path": date_path,  # 传入计算好的日期路径
        }
    )

此时query.sql中直接使用参数即可：

{{ params.date_path }}

方法3：使用动态任务映射（Airflow 2.2+） #

如果你的Airflow版本是2.2及以上，推荐用动态任务映射简化循环创建任务的逻辑：

from airflow.decorators import dag, task
from airflow.providers.google.cloud.operators.bigquery import BigQueryOperator
from datetime import datetime

@dag(start_date=datetime(2024, 5, 20), schedule_interval="@daily")
def my_dag():
    bq_task = BigQueryOperator.partial(
        task_id="bq_to_gcs",
        sql=path.join("queries", "query.sql"),
        params={"schema": BQ_SOURCE_DATA_SET_ID},
    ).expand(
        params=[{"days": days} for days in range(0, 3)]
    )

my_dag()

SQL模板仍使用方法1中的写法即可。

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