Airflow技术问题：回推数据至管道失败，XCOM推送报错返回码-9

报错原因 #

INFO - Task exited with return code -9 并非XCOM推送功能本身的问题，而是系统因内存不足强制终止了第二个任务（return code -9对应Linux系统的SIGKILL信号，通常由OOM Killer触发）。任务被强制终止后，无法完成XCOM数据推送的收尾操作，才会出现看似XCOM失效的现象。

解决步骤 #

1. 排查任务内存消耗 #

检查第二个任务的数据处理逻辑：

• 是否一次性加载了超大数据集到内存？
• 是否存在内存泄漏（比如未及时清理大变量、循环中持续累积数据）？
• 是否执行了高内存开销的计算（比如无限制的聚合、排序操作）？

2. 优化数据处理逻辑 #

• 改用分块处理：避免一次性加载全量数据，按批次处理后及时释放内存
• 替换大对象存储：如果数据量超过几MB，不要直接用XCOM存储原始数据，而是将数据写入外部存储（如S3、数据库），XCOM仅存储数据的访问路径
• 清理无用资源：手动删除不再使用的变量、关闭文件句柄，避免内存占用持续上升

3. 调整Airflow资源配置 #

• 如果使用Kubernetes Executor，在任务定义中通过resources参数指定内存请求和限制：

  PythonOperator(
      task_id="task2",
      python_callable=your_process_function,
      resources={
          'request_memory': '2Gi',
          'limit_memory': '4Gi'
      }
  )
  

• 如果使用Celery Executor，调整worker节点的内存配额，避免单个任务占用过多资源

4. 确认XCOM的正确使用方式 #

任务间数据通信的标准XCOM用法：

• 推送数据：在任务函数中通过任务实例（ti）主动推送，或通过返回值自动推送

  def task1_process(ti):
      processed_data = {"result": "sample_data"}
      # 方式1：主动推送指定key的数据
      ti.xcom_push(key="task1_output", value=processed_data)
      # 方式2：返回值自动推送到key='return_value'
      return processed_data
  

• 拉取数据：在后续任务中拉取指定任务的XCOM数据

  def task2_process(ti):
      # 拉取task1的return_value
      default_data = ti.xcom_pull(task_ids="task1")
      # 拉取指定key的数据
      specific_data = ti.xcom_pull(task_ids="task1", key="task1_output")
  

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