如何在Airflow DAG中动态修改存储桶名称？

要实现通过Airflow变量动态指定存储桶名称读取SQL文件，核心是区分DAG解析阶段和任务执行阶段的代码执行时机，避免在解析阶段尝试渲染模板变量。以下是三种可行方案：

方案1：使用PythonOperator读取SQL并通过XCom传递（推荐） #

这种方法将文件读取逻辑放到任务执行阶段，确保每次任务运行时获取最新的Airflow变量值，灵活性最高：

from airflow.models import Variable
from airflow.operators.python import PythonOperator
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator

with DAG(
    "my_dag",
    default_args=your_default_args,
    schedule_interval=None,
) as dag:
    def fetch_sql_file(file_name, **context):
        # 任务执行时获取最新变量值
        bucket_name = Variable.get("my_dynamic_bucket")
        file_path = f"/home/airflow/gcs/data/{bucket_name}/{file_name}"
        with open(file_path, "r") as f:
            sql_content = f.read()
        # 将SQL内容推送到XCom，供后续任务调用
        context["ti"].xcom_push(key="sql_query", value=sql_content)

    # 定义读取SQL的任务
    fetch_sql_task = PythonOperator(
        task_id="fetch_sql_from_bucket",
        python_callable=fetch_sql_file,
        op_kwargs={"file_name": "my_sql_file.sql"},
        provide_context=True,
    )

    # 定义BigQuery执行任务，从XCom拉取SQL内容
    run_bq_job = BigQueryInsertJobOperator(
        task_id="execute_bigquery_job",
        configuration={
            "query": {
                "query": "{{ ti.xcom_pull(key='sql_query', task_ids='fetch_sql_from_bucket') }}",
                "useLegacySql": False,
                # 补充你的其他BigQuery配置
            }
        },
    )

    # 设置任务依赖
    fetch_sql_task >> run_bq_job

方案2：利用jinja模板直接读取文件 #

如果你的Airflow环境允许使用jinja的open函数，可以直接在BigQueryInsertJobOperator的配置中渲染路径并读取文件，代码更简洁：

from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator

with DAG(
    "my_dag",
    default_args=your_default_args,
) as dag:
    run_bq_job = BigQueryInsertJobOperator(
        task_id="execute_bigquery_job",
        configuration={
            "query": {
                # 通过jinja模板拼接路径并读取文件内容
                "query": "{{ open('/home/airflow/gcs/data/' + var.value.my_dynamic_bucket + '/my_sql_file.sql').read() }}",
                "useLegacySql": False,
                # 补充你的其他BigQuery配置
            }
        },
        # 确保configuration字段支持模板渲染（Airflow 2.x默认已支持）
        template_fields=["configuration"],
    )

方案3：DAG解析阶段读取变量（适合变量不频繁变更场景） #

如果存储桶名称很少变化，可以在DAG解析时直接读取变量，代码最简单，但变量更新后需要重启Airflow调度器或触发DAG重新解析才能生效：

from airflow.models import Variable
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator

# DAG解析时读取变量
bucket_name = Variable.get("my_dynamic_bucket")

with DAG(
    "my_dag",
    default_args=your_default_args,
) as dag:
    def get_query(file_name):
        file_path = f"/home/airflow/gcs/data/{bucket_name}/{file_name}"
        with open(file_path) as f:
            return f.read()

    run_bq_job = BigQueryInsertJobOperator(
        task_id="execute_bigquery_job",
        configuration={
            "query": {
                "query": get_query("my_sql_file.sql"),
                "useLegacySql": False,
                # 补充你的其他BigQuery配置
            }
        },
    )

你之前尝试失败的原因 #

1. Python函数内使用模板字符串：Airflow不会渲染Python函数内部的{{ var.value... }}模板，这些字符串会被当作普通路径的一部分，导致文件找不到。
2. 直接在configuration中执行open：open()函数在DAG解析阶段就会被执行，此时模板变量还未渲染，Python解析器无法识别{{ }}语法，引发错误。

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