Airflow中BashOperator调用的Python文件如何使用xcom_push与xcom_pull

你不能直接在被BashOperator调用的独立Python脚本里调用xcom_push和xcom_pull方法。
原因是：XCom是Airflow运行时上下文中的对象，只有在Airflow Worker调度的任务进程内才能访问对应的操作句柄。你通过bash命令启动的my_task.py是完全独立的子进程，没有加载Airflow的任务上下文，自然无法直接调用XCom相关接口。

下面给出两种可行的实现方案：

方案1：继续使用BashOperator，通过传参和标准输出实现XCom读写 #

读取XCom #

你可以利用Airflow的模板语法，在bash_command中提前把需要拉取的XCom值作为参数传给Python脚本：

my_task = BashOperator(
    task_id='my_task',
    bash_command='python3 /opt/airflow/dags/programs/my_task.py {{ ti.xcom_pull(task_ids="你要拉取的上游任务ID") }}',
    do_xcom_push=True
)

之后在my_task.py中直接读取命令行参数即可拿到XCom值：

import sys
if __name__ == "__main__":
    pulled_xcom_value = sys.argv[1]
    # 后续业务逻辑

推送XCom #

你只需要在my_task.py的末尾，把需要推送的内容通过print()输出到标准输出即可。BashOperator开启do_xcom_push=True后，会自动把脚本的标准输出内容推送到XCom中，下游任务直接拉取当前my_task的XCom就能拿到对应值。

# my_task.py 末尾示例
push_value = "需要推送到XCom的内容"
print(push_value)

方案2：改用PythonOperator，直接操作XCom #

如果你不想通过传参和打印的方式中转，直接把my_task.py中的逻辑封装成可导入的函数，在DAG中用PythonOperator调用，就可以直接在函数内操作XCom：

# 首先导入你脚本里的逻辑函数
from programs.my_task import your_logic_func
from airflow.operators.python import PythonOperator

def my_task_func(**context):
    ti = context['ti']
    # 直接拉取XCom
    pulled_value = ti.xcom_pull(task_ids="上游任务ID")
    result = your_logic_func(pulled_value)
    # 直接推送XCom
    ti.xcom_push(key="自定义key", value=result)

my_task = PythonOperator(
    task_id='my_task',
    python_callable=my_task_func,
    provide_context=True
)

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