Airflow中BashOperator的xcom_pull返回None，调度运行却正常

核心原因 #

你测试时ti.xcom_pull()返回None，是因为测试场景没有正确生成Airflow任务实例的上下文，也没有实际执行get_datetime任务并存储XCom数据；而调度运行时，Airflow会完整执行任务链，自动处理XCom的存储与读取，所以能正常拿到值。

解决方案 #

1. 正确的任务测试方式

不要直接调用process_datetime函数，而是用Airflow官方的测试命令来模拟任务执行，确保XCom数据被正确存储：

• 先执行get_datetime任务，生成XCom数据：

  airflow tasks test first_airflow_dag get_datetime 2022-02-01T00:00:00
  

• 再执行process_datetime任务：

  airflow tasks test first_airflow_dag process_datetime 2022-02-01T00:00:00
  

  注：最后的时间参数需与DAG的start_date或有效执行日期一致。

2. 修复代码中的XCom处理逻辑

BashOperator默认会捕获命令输出的最后一行作为XCom返回值，类型是字符串而非列表。你代码中dt[0]的写法存在逻辑错误，会导致调度时实际取的是字符串的第一个字符（而非完整日期），只是未触发异常而已。修改后的函数如下：

def process_datetime(ti):
    dt = ti.xcom_pull(task_ids='get_datetime')
    if not dt:
        raise Exception('No datetime value.')

    # 直接对字符串进行split，无需取dt[0]
    dt_parts = dt.split()
    return dt_parts

3. 可选：显式指定BashOperator的XCom返回规则

如果需要捕获命令的全部输出（而非最后一行），可以显式配置相关参数：

task_get_datetime = BashOperator(
    task_id='get_datetime',
    bash_command='date',
    # 显式开启XCom推送（默认已开启，可省略）
    do_xcom_push=True
)

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