Airflow中获取XCOM返回None值问题排查

你的代码核心问题是XCOM拉取时指定的task_id与实际任务ID不匹配，导致无法获取到值，返回None。

问题分析 #

你定义的task3的task_id是XCOM_Example，但在get_name_xcom函数中调用ti.xcom_pull(task_ids='get_name')，这个get_name并非任何任务的task_id，自然拉取不到数据。

修正方案 #

将ti.xcom_pull中的task_id改为实际任务的IDXCOM_Example，修正后的代码如下：

def get_name_xcom(ti):
    # 匹配task3的task_id
    name_xcom = ti.xcom_pull(task_ids='XCOM_Example')
    print(f"Hi {name_xcom}. This was pulled from XCOM.")

# 任务定义保持不变
task3 = PythonOperator(
    task_id='XCOM_Example',
    python_callable=get_name
)

task4 = PythonOperator(
    task_id='Greet_with_XCOM',
    python_callable=get_name_xcom
)

task3 >> task4

其他注意事项 #

• 若手动使用ti.xcom.push写入XCOM，需保证拉取时指定的key与写入时一致（默认key为return_value，无需手动指定）：

  def get_name(ti):
      ti.xcom_push(key='custom_name', value='Ashish')
  
  def get_name_xcom(ti):
      name_xcom = ti.xcom_pull(task_ids='XCOM_Example', key='custom_name')
      print(f"Hi {name_xcom}. This was pulled from XCOM.")
  

• 确认任务依赖正确：task3 >> task4确保task3执行完成后才运行task4，保证XCOM数据已生成。
• 可通过Airflow UI的XCOM面板，查看对应任务的XCOM数据是否存在，验证写入是否成功。

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