Apache Airflow中xcom_pull报'task_id'参数错误的问题与解决

我正在学习Marc Lamberti的Apache Airflow课程，在名为process_user的任务中遇到问题，相关代码如下：

def _process_user(ti):
    user = ti.xcom_pull(task_id = "extract_user")
    user = user['results'][0]
    processed_user =  json_normalize({
        'first_name': user['name']['first'],
        'last_name': user['name']['last'],
        'country': user['location']['country'],
        'username': user['login']['username'],
        'password': user['login']['password'],
        'email': user['email']})
    processed_user.to_csv('/tmp/processed_user.csv', index=None, header=False)

...

with DAG(dag_id='user_processing',
        start_date=datetime(2022, 1, 1),
        schedule_interval='@daily',
        catchup=False) as dag:

    ...

    extract_user = SimpleHttpOperator(
        task_id='extract_user',
        http_conn_id='user_api',
        endpoint='api/',
        method='GET',
        response_filter = lambda response: 
        json.loads(response.text),
        log_response = True
    )

    process_user = PythonOperator(
        task_id='process_user',
        python_callable = _process_user
    )

    ...

运行后持续报错：

TypeError: xcom_pull() got an unexpected keyword argument 'task_id'

错误原因 #

两个关键点导致报错：

1. xcom_pull方法的参数名称是**task_ids（复数形式）**，而非代码中使用的task_id（单数），属于参数名拼写错误。
2. 直接在函数参数中声明ti的写法，在部分Airflow版本中需要配合PythonOperator设置provide_context=True才能正常传递上下文对象；改用**kwargs接收上下文是更通用且稳妥的方式。

解决方案 #

修改_process_user函数的定义，通过**kwargs获取上下文对象，同时修正xcom_pull的参数名：

def _process_user(**kwargs):
    user = kwargs['ti'].xcom_pull(task_ids="extract_user")
    # 后续处理逻辑保持不变
    user = user['results'][0]
    processed_user =  json_normalize({
        'first_name': user['name']['first'],
        'last_name': user['name']['last'],
        'country': user['location']['country'],
        'username': user['login']['username'],
        'password': user['login']['password'],
        'email': user['email']})
    processed_user.to_csv('/tmp/processed_user.csv', index=None, header=False)

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