Airflow 2.0中如何在DAG构造函数内访问触发配置变量设置dag_id

你当前的写法是不可行的，核心原因是dag_id是DAG本身的唯一标识，由Airflow调度器定期解析dag.py文件时就确定生成；而触发DAG时传入的conf参数属于**单次DAG运行（DAG Run）**的私有参数，两个操作的时间点、作用域完全不同，DAG初始化阶段无法访问还没产生的运行时参数。

以下为对应场景的可行方案：

场景1：需要管理多版本DAG，不同版本对应独立的DAG实例 #

你可以直接在DAG定义文件中指定版本号，或从Airflow全局变量中读取版本来生成dag_id，示例代码如下：

DAG定义示例 #

from airflow.models import Variable
from airflow import DAG
from datetime import datetime

# 方式1：硬编码版本
# DAG_VERSION = "some_dag_v1.0.0"
# 方式2：从Airflow Variables读取版本，可在Airflow UI的Admin-Variables页面修改
DAG_VERSION = Variable.get("some_dag_version", default_var="some_dag_v1.0.0")
# 方式3：如果需要批量生成多版本DAG，可直接遍历版本列表循环生成
# VERSIONS = ["v1.0.0", "v1.1.0", "v2.0.0"]
# for ver in VERSIONS:
#     with DAG(dag_id=f"some_dag_{ver}", ...) as dag:
#         # 定义任务逻辑
#         globals()[f"some_dag_{ver}"] = dag

dag = DAG(
    dag_id=f"some_dag_{DAG_VERSION}",
    schedule_interval="@daily",
    start_date=datetime(2023, 1, 1),
    catchup=False
)

触发脚本不需要额外传版本参数，直接指定对应版本的dag_id即可：

DAG_VERSION=some_dag_v1.0.0
gcloud beta composer environments run "${AIRFLOW_ENV_NAME}" --location=us-central1 dags trigger -- "some_dag_${DAG_VERSION}"

场景2：只需要给单次DAG运行传版本参数，不需要修改dag_id #

如果版本参数只用于单次运行的业务逻辑，不需要生成独立DAG，那么固定dag_id即可，在任务内部读取触发传入的conf参数使用：

触发脚本调整 #

DAG_VERSION=some_dag_v1.0.0
TRIGGER_PARAMS='{"dag_version":"'"${DAG_VERSION}"'"}';
# 此处指定固定的dag_id，而非版本号
gcloud beta composer environments run "${AIRFLOW_ENV_NAME}" --location=us-central1 dags trigger -- "fixed_some_dag_id" --conf "${TRIGGER_PARAMS}";

DAG定义示例 #

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def process_logic(**context):
    # 运行时读取触发传入的参数
    run_conf = context['dag_run'].conf
    dag_version = run_conf.get("dag_version")
    # 后续用dag_version执行业务逻辑
    print(f"当前运行的DAG版本为: {dag_version}")

with DAG(
    dag_id="fixed_some_dag_id",
    schedule_interval=None,
    start_date=datetime(2023, 1, 1),
    catchup=False
) as dag:
    process_task = PythonOperator(
        task_id="process_task",
        python_callable=process_logic,
        provide_context=True
    )

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