手动触发Airflow DAG时，能否强制用户指定run_id？

Airflow 本身不支持直接强制用户在触发时指定 run_id，但可以通过自定义必填参数验证+关联 run_id的方式实现你的需求——确保手动触发时必须提供数据转储ID，同时将其与 run_id 绑定。

方法一：强制用户输入转储ID，并关联到 run_id #

1. 定义DAG时设置必填参数
   在DAG定义中通过 params 指定必填的 dump_id，Airflow UI 触发时会自动生成表单要求用户填写：

   from airflow import DAG
   from airflow.operators.python import PythonOperator
   from datetime import datetime
   
   def validate_dump_id(**context):
       dump_id = context['dag_run'].conf.get('dump_id')
       if not dump_id:
           raise ValueError("必须指定数据转储ID（dump_id）")
       # 将dump_id与run_id绑定，若dump_id存在重复风险，可拼接时间戳保证唯一性
       context['dag_run'].run_id = f"dump_{dump_id}_{datetime.now().strftime('%Y%m%d%H%M%S')}"
   
   with DAG(
       dag_id='data_dump_processor',
       start_date=datetime(2023, 1, 1),
       params={
           "dump_id": {
               "type": "string",
               "required": True,
               "description": "数据转储的唯一ID"
           }
       },
       catchup=False
   ) as dag:
       validate_task = PythonOperator(
           task_id='validate_dump_id',
           python_callable=validate_dump_id,
           provide_context=True
       )
       # 后续数据处理任务示例
       # process_task = PythonOperator(...)
       validate_task >> process_task
   

   用户在UI手动触发时，必须填写 dump_id 才能提交表单；验证任务会将 run_id 设置为包含 dump_id 的值，实现两者关联，同时保证 run_id 唯一性。

2. API触发时的强制验证
   若通过Airflow API触发DAG，需在请求的 conf 参数中携带 dump_id，验证任务会自动检查参数是否存在，不存在则直接终止DAG运行。

方法二：直接强制指定 run_id（规范+验证） #

• UI层面：Airflow手动触发页面允许用户手动输入 run_id，可通过团队内部规范要求用户将 dump_id 作为 run_id 的值，再在DAG开头添加验证任务，检查 run_id 是否符合指定格式（比如是否以 dump_ 开头），不符合则终止DAG。
• API层面：封装内部触发接口，强制调用方传入 dump_id，并将其直接设置为 run_id 后再调用Airflow官方API，从源头确保 run_id 与 dump_id 绑定。

关键注意事项 #

• run_id 必须全局唯一，若你的 dump_id 本身是全局唯一的，可直接用它作为 run_id；若存在重复风险，务必拼接时间戳或其他唯一标识。
• 验证任务要放在DAG的最开头，避免参数不合法时执行后续任务造成资源浪费。

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