如何通过字典向Airflow Operator动态传递参数

该需求完全可实现，直接使用Python原生的*关键字参数解包（** 操作符）*即可满足，不需要依赖Airflow的特殊参数。

基础实现 #

你之前的写法中把字典直接赋值给arguments参数是错误的，正确用法是将字典通过**解包为Operator的关键字入参：

arguments = {
    "task_id": "Bash_task",
    "bash_command": "echo \"here is the message: '$message'\"",
    # 可自由添加任意Operator支持的参数
    "env": {"message": '{{ dag_run.conf["message"] if dag_run else "" }}'},
    "retries": 2
}

# 字典解包后传入
bash_task = BashOperator(**arguments)

动态参数构造示例 #

你可以先构造基础参数字典，再根据业务条件动态添加、修改参数，完全不需要硬编码参数名和取值：

# 初始化基础参数
base_args = {
    "task_id": "dynamic_bash_task",
    "retries": 1
}

# 按条件动态调整参数
if use_custom_env:
    base_args["env"] = {"message": '{{ dag_run.conf["message"] if dag_run else "" }}'}
if run_prod_script:
    base_args["bash_command"] = "sh /opt/prod/task_script.sh"
else:
    base_args["bash_command"] = "echo 'test run'"
base_args["execution_timeout"] = timedelta(minutes=30 if is_long_task else 5)

# 实例化Operator
bash_task = BashOperator(**base_args)

结合Yaml配置的实现 #

如果你已经通过yaml文件传递参数值，直接读取yaml文件得到字典后解包即可：

import yaml

# 读取yaml配置
with open("/path/to/operator_config.yaml", "r", encoding="utf-8") as f:
    operator_config = yaml.safe_load(f)

# 直接用yaml配置生成Operator
bash_task = BashOperator(**operator_config)

补充说明 #

你之前使用params参数不符合预期是正常的：params是Airflow用于向任务运行上下文传递自定义变量的参数，作用是给任务运行阶段传值，不是给Operator构造阶段传入参的，两者定位完全不同。

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