Airflow中多函数共享params及后续任务配置问题（含KeyError报错）

问题场景 #

现有一段使用params的Airflow代码，需要实现以下需求：

• 在my_main函数中调用anotherfunction、someotherfunction，让这两个函数也能访问DAG定义的params
• 创建finalfunction作为my_main执行成功后的后续任务，该任务同样需要读取params

运行时触发报错：

params = context['params']
KeyError: 'params'

错误原因 #

1. 调用anotherfunction()和someotherfunction()时未传递上下文（context）或params参数，导致函数内部无法获取context['params']
2. finalfunction未被定义为Airflow任务，无法自动获取上下文
3. 原代码通过op_kwargs传递模板化参数的方式冗余，未充分利用Airflow上下文直接获取params的能力

修复后的完整代码 #

from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator
import ast 

def my_main(**kwargs):
    # 直接从context中获取params
    params = kwargs['params']
    print(params['is_debug'])  # bool类型无需转换
    print(params['lst'])       # list类型无需转换

    # 调用子函数时传递上下文
    anotherfunction(**kwargs)
    someotherfunction(**kwargs)

def anotherfunction(**context):
    params = context['params']
    print(params['is_debug'])

def someotherfunction(**context):
    params = context['params']
    print(params['is_debug'])

def finalfunction(**context):
    params = context['params']
    print(params['is_debug'])

with DAG(
        dag_id="my_main",
        description='my_main',
        start_date=datetime(2022, 6, 8, 1, 0),
        schedule_interval=None,
        catchup=False,
        params={
            "is_debug": False,
            "lst": ["a", "b"],
        },
) as dag:
    task_my_main = PythonOperator(
        task_id='task_my_main_main',
        provide_context=True,
        python_callable=my_main,
        # 无需单独传递op_kwargs，直接通过context获取params
    )

    # 添加后续任务finalfunction
    task_final = PythonOperator(
        task_id='task_final',
        provide_context=True,
        python_callable=finalfunction,
    )

    # 设置任务依赖：task_my_main执行成功后再运行task_final
    task_my_main >> task_final

关键修改说明 #

• 传递上下文给子函数：在my_main中调用子函数时，通过anotherfunction(**kwargs)将上下文完整传递，确保子函数能读取context['params']
• 定义后续任务：将finalfunction封装为PythonOperator任务，开启provide_context=True，让任务自动获取DAG上下文及params
• 简化参数读取：直接从上下文的params字段读取配置，删除冗余的模板变量（is_debug_param等）和op_kwargs，代码更简洁高效

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