Airflow UI参数无法传递至DAG问题求助

核心问题 #

你的代码中直接将定义阶段的Param对象传递给op_kwargs，这会导致任务始终使用Param的默认值，不会读取用户在UI触发时输入的参数。因为my_params里的v是Param实例，不是运行时的实际参数值。

修复代码 #

方法1：使用Jinja模板引用运行时参数 #

修改PythonOperator的op_kwargs，通过Airflow的Jinja模板语法获取实际参数：

my_params = {
    "schema_prefix": Param(
        "ABC",
        description="Prefix to schema",
        type="string",
        minLength=2)
}

with DAG(
    dag_id="test_dag_1",
    start_date=datetime(2024, 1, 1),
    params=my_params,
) as dag:
    python_task = PythonOperator(
            python_callable=my_task_python_file,
            task_id="python_task_id",
            op_kwargs={
                "schema_prefix": "{{ params.schema_prefix }}"
            },
        )

方法2：通过函数上下文获取参数 #

如果你的Python可调用函数接受context参数，可以直接从上下文中读取params：

def my_task_python_file(**context):
    schema_prefix = context["params"]["schema_prefix"]
    # 后续逻辑...

# DAG定义部分
my_params = {
    "schema_prefix": Param(
        "ABC",
        description="Prefix to schema",
        type="string",
        minLength=2)
}

with DAG(
    dag_id="test_dag_1",
    start_date=datetime(2024, 1, 1),
    params=my_params,
) as dag:
    python_task = PythonOperator(
            python_callable=my_task_python_file,
            task_id="python_task_id",
            provide_context=True,  # Airflow 2.x中可省略，默认已开启
        )

环境变量配置确认 #

确保AIRFLOW__CORE__DAG_RUN_CONF_OVERRIDES_PARAMS=true已正确生效：

• 检查Airflow配置文件airflow.cfg中是否存在该配置且值为true
• 修改配置后需重启Airflow Webserver和Scheduler服务

UI传参最佳实践 #

• 触发DAG时，在UI的**"Params"**区域（Airflow 2.x及以上版本）直接输入参数值，无需手动写入JSON格式
• 若使用旧版本Airflow，在**"Conf"**区域以JSON格式传递参数：{"schema_prefix": "你的自定义值"}
• 避免使用Airflow全局变量，因为全局变量是共享的，无法满足不同用户触发时使用不同参数的需求

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