如何从Airflow UI参数化DAG，手动触发时指定org参数？

兄弟，听我一句劝，别生成N个DAG！这绝对是给自己挖维护大坑——以后改逻辑要改N遍，UI里一堆重复的DAG看着都闹心。直接用Airflow的参数传递功能就能完美解决手动触发时选org的需求，简单又优雅！

核心思路是用一个通用DAG，把org作为可配置参数，手动触发时指定具体值就行。

方法1：用DAG的params定义默认参数 #

直接在DAG定义里声明params，给个默认值，然后任务中读取这个参数。代码示例：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import myapi  # 导入你的API模块

def run_compute_metrics(**context):
    # 从上下文获取手动触发时传入的org参数
    target_org = context["params"]["org"]
    print(f"开始处理 org: {target_org}")
    myapi.compute_metrics(target_org)
    print(f"完成 org: {target_org} 的指标计算")

# 定义通用DAG
with DAG(
    dag_id="compute_entity_metrics",
    schedule_interval=None,  # 临时运行，所以关闭自动调度
    start_date=datetime(2024, 1, 1),
    catchup=False,
    params={
        "org": "default_org"  # 设置默认值，手动触发时可修改
    }
) as dag:
    compute_task = PythonOperator(
        task_id="compute_metrics_task",
        python_callable=run_compute_metrics,
        provide_context=True  # 必须开启，才能在任务中获取params
    )

手动触发时怎么选org？ #

在Airflow UI找到这个DAG，点击右上角的 Trigger DAG w/ config，在弹出的配置框里，修改params中的org值（比如把"default_org"改成"your_target_org"），然后点击触发就可以了。

方法2：用dag_run.conf传递参数（更灵活） #

如果不想依赖DAG的默认params，也可以通过DAG Run的配置直接传参，代码里读取dag_run.conf：

def run_compute_metrics(**context):
    # 从dag_run的配置中获取org
    target_org = context["dag_run"].conf.get("org")
    if not target_org:
        raise ValueError("必须指定org参数！")
    myapi.compute_metrics(target_org)

# DAG定义不需要params，保持简洁
with DAG(
    dag_id="compute_entity_metrics",
    schedule_interval=None,
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag:
    compute_task = PythonOperator(
        task_id="compute_metrics_task",
        python_callable=run_compute_metrics,
        provide_context=True
    )

手动触发时，在配置框里输入JSON格式的参数：{"org": "your_target_org"}即可。

进阶：给org加合法性校验 #

如果你的org列表是固定的，可以用Airflow Variable存储允许的org列表，在任务里做校验，防止输入错误：

from airflow.models import Variable

def run_compute_metrics(**context):
    target_org = context["params"]["org"]
    # 从Variable获取允许的org列表（先在Airflow UI的Variables里创建这个变量，值是JSON数组）
    allowed_orgs = Variable.get("allowed_orgs", deserialize_json=True)
    if target_org not in allowed_orgs:
        raise ValueError(f"非法的org！允许的列表：{allowed_orgs}")
    myapi.compute_metrics(target_org)

• 维护成本爆炸：以后要修改compute_metrics的逻辑、调整任务依赖，你得改N个DAG，想想都头大；
• UI混乱：一堆名字类似的DAG占满列表，找起来麻烦；
• 资源浪费：每个DAG都会占用Airflow的调度资源，完全没必要。

这个单DAG+参数的方式，既满足手动选org的需求，又能保持代码的简洁性和可维护性，完美适配你的临时运行场景！

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