如何为Airflow DAG添加参数并设置默认值及GUI触发配置

一、代码修改实现默认参数 #

你需要在Python调用函数中从Airflow上下文获取运行参数，并设置默认值，具体修改如下：

修改后的完整代码 #

import lib.log as log
import traceback
import datetime
import json

def my_main(**kwargs):
    try:
        # 从DAG运行上下文获取配置参数，未指定时使用默认值
        dag_conf = kwargs.get('dag_run', {}).conf or {}
        is_debug = dag_conf.get('is_debug', False)
        seti = dag_conf.get('seti', 'FG')

        if is_debug:
            log.log_it('U in debug mode')
        else:
            log.log_it('U in normal mode')
        
        if seti == 'FG':
            log.log_it('FG')
        else:
            log.log_it('RX')
    except Exception:
        traceback.print_exc()
        raise  # 抛出异常让Airflow标记任务失败

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator


dag = DAG(
    'my_main', 
    description='my_main', 
    start_date=datetime.datetime(2023,6,8,1,0), 
    schedule_interval="0 1 * * *",
    catchup=False
)
task_my_main = PythonOperator(
    task_id='task_my_main_main', 
    provide_context=True, 
    python_callable=my_main, 
    dag=dag
)

关键修改点说明 #

• 通过kwargs.get('dag_run', {}).conf获取手动触发时传入的配置参数，未传入则使用空字典
• 用dict.get(key, default)方法设置默认值：is_debug默认False，seti默认'FG'
• 补全异常处理逻辑，确保任务异常时能被Airflow正确捕获并标记失败

二、GUI手动触发时的参数填写方法 #

在Airflow UI手动触发DAG时，找到「Configuration JSON (Optional): {}」输入框，按照JSON格式填写参数：

示例1：同时修改两个参数 #

{
  "is_debug": true,
  "seti": "RX"
}

示例2：仅修改单个参数（另一个沿用默认值） #

{
  "is_debug": true
}

示例3：完全使用默认值 #

保持输入框默认的{}即可，或直接留空。

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