Airflow中BashOperator无法传递XCom至其他Operator的解决方法

核心问题原因 #

BashOperator的do_xcom_push=True默认只会捕获**任务最后一行的标准输出(stdout)**作为XCom值。如果第一个Python脚本没有正确输出目标变量到stdout，或者输出被缓冲、不是最后一行，就会导致XCom中找不到对应值。

具体解决步骤 #

1. 修正第一个Python脚本(myscript.py) #

确保要传递的字符串是脚本的最后一行输出，且直接打印到stdout：

# myscript.py示例代码
# 处理逻辑...
target_variable = "要传递的字符串内容"
# 仅在最后一行输出目标变量，不要添加多余的print或日志
print(target_variable)

2. 优化第一个BashOperator的执行命令 #

添加Python的-u参数关闭输出缓冲，确保脚本的stdout实时输出被Airflow捕获：

first_status = BashOperator(
    task_id=1st_taskid_str,
    # 添加-u参数关闭缓冲
    bash_command=f"python -u myscript.py --dag_id '{dag.dag_id}' \
                            --task_id '{1st_taskid_str}' --dag_conf configuration_string",
    retries=10,
    dag=dag,
    retry_delay=timedelta(minutes=1),
    do_xcom_push=True,
)

3. 修正第二个BashOperator的XCom引用语法 #

简化模板变量的引号写法，避免嵌套转义出错：

second_status = BashOperator(
    task_id=process_file_task_id,
    # 用单引号包裹XCom模板，避免双引号转义问题
    bash_command=f"python secondscript.py --dag_id '{dag.dag_id}' \
                            --task_id '{2nd_task_id}' --dag_conf configuration_string --file '{{{{ ti.xcom_pull(task_ids=\"{1st_taskid_str}\") }}}}'",
    dag=dag,
    # 显式声明模板字段，确保Airflow解析模板
    template_fields=('bash_command',)
)

注意：Airflow模板需要双层大括号{{{ ... }}}，因为外层的f-string会先解析一层{}，最终传递给Airflow的是{{ ti.xcom_pull(...) }}。

4. 验证XCom是否正常生成 #

运行第一个任务后，在Airflow UI的XComs页面，筛选对应DAG和1st_taskid_str任务，检查是否存在key为return_value的XCom记录，其值应为脚本输出的目标字符串。

额外注意事项 #

• 确保第一个脚本没有将输出写入stderr（比如用sys.stderr.write()），否则不会被BashOperator捕获。
• 如果脚本有大量日志输出，建议将日志重定向到文件，只保留目标变量在最后一行stdout：

  python -u myscript.py ... > /tmp/myscript.log 2>&1 && tail -n 1 /tmp/myscript.log
  

  这种写法会把所有输出存入日志文件，仅将最后一行（目标变量）输出到stdout供XCom捕获。

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