手动触发Airflow DAG时如何给PythonOperator传递可选参数？

当你通过「Trigger DAG w/ config」手动触发DAG，使用dag_run.conf['param2']访问可选参数时，若未传入该参数，Jinja2模板会因为字典中不存在对应的key抛出UndefinedError。解决这个问题的核心是避免直接用下标访问可选参数，改用以下两种方式处理：

方法1：使用字典get()方法设置默认值 #

dict.get(key, default)方法允许你指定当key不存在时返回的默认值，既可以设为具体值，也可以设为None（未传递时默认返回None）。

修改参数定义部分：

# Parameters
param1="{{ dag_run.conf['param1'] }}"  # 必填参数，确保触发时必须传入
param2="{{ dag_run.conf.get('param2', None) }}"  # 可选参数，未传递时返回None

如果需要给param2设置具体默认值（比如空字符串），可改为：

param2="{{ dag_run.conf.get('param2', '') }}"

方法2：使用Jinja2条件判断 #

通过条件判断检查key是否存在，不存在时返回默认值，适合需要更复杂逻辑的场景：

# Parameters
param1="{{ dag_run.conf['param1'] }}"
param2="{{ dag_run.conf['param2'] if 'param2' in dag_run.conf else None }}"

额外注意事项 #

确保你的Python可调用函数（my_function.main）能处理param2为None或默认值的情况，比如给函数参数设置默认值：

# 在my_function.py中
def main(param1, param2=None):
    # 业务逻辑处理
    if param2 is not None:
        # 处理param2存在的情况
        pass
    # 其他逻辑

修改后的完整DAG代码示例：

__version__ = "@version@"

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.trigger_rule import TriggerRule
import sys
import socket
import datetime
import my_function

file_name = my_function.get_file_name(__file__)

dag = DAG(
    dag_id = file_name,
    default_args=my_function.default_args,
    catchup=False,
    schedule_interval=None,
    tags = ['my_function']
)

# Parameters
param1="{{ dag_run.conf['param1'] }}"
param2="{{ dag_run.conf.get('param2', None) }}"

custom_task = PythonOperator(
    task_id = "custom_task",
    dag=dag,
    python_callable=my_function.main,
    op_kwargs={'param1': param1, 'param2': param2},
    trigger_rule=TriggerRule.ALL_SUCCESS
)

custom_task

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