Airflow中如何获取PythonOperator返回值并传递给后续任务？

嘿，这个问题我太熟了！在Airflow里，要让一个PythonOperator的返回值被下一个任务拿到，核心就是用XCom——Airflow内置的任务间数据共享机制，其实你的代码已经迈出第一步了，咱们一步步来：

第一步：你的Task1已经自动把返回值存进XCom了 #

默认情况下，PythonOperator执行的函数如果返回了非None的值，Airflow会自动把这个值推送到XCom中，对应的key是return_value，关联的task_id就是你的Data_Extraction_Environment。所以你现在的Task1代码完全不用改，它已经在默默帮你存数据了：

task1 = af_op.PythonOperator(task_id='Data_Extraction_Environment', provide_context=True, python_callable=Task1, dag=dag1)
def Task1(**kwargs):
    return(kwargs['dag_run'].conf.get('file'))

第二步：在下一个PythonOperator里拉取这个值 #

接下来要写第二个任务，通过TaskInstance（简称ti）的xcom_pull()方法来获取之前存的值。因为你设置了provide_context=True，所以任务函数的kwargs参数里会包含ti对象，直接用就行：

def Task2(**kwargs):
    # 从XCom拉取task1的返回值，指定对应的task_id即可
    file_path = kwargs['ti'].xcom_pull(task_ids='Data_Extraction_Environment')
    # 这里就可以用拿到的值做后续处理了
    print(f"成功获取到任务1传递的文件路径：{file_path}")
    # 比如调用自定义的文件处理逻辑
    # process_target_file(file_path)

# 定义第二个PythonOperator
task2 = af_op.PythonOperator(
    task_id='Process_Extracted_Data',
    provide_context=True,
    python_callable=Task2,
    dag=dag1
)

# 设置任务依赖，确保task1执行完成后再启动task2
task1 >> task2

额外技巧：自定义XCom的key #

如果你不想用默认的return_value，也可以在Task1里主动推送XCom，指定自定义的key，这样代码可读性更强：

def Task1(**kwargs):
    file_val = kwargs['dag_run'].conf.get('file')
    # 主动推送XCom，指定key为'input_file_path'
    kwargs['ti'].xcom_push(key='input_file_path', value=file_val)
    return file_val  # 这里返回的值仍会存到默认的return_value，主动推送更灵活

# 然后在Task2里拉取的时候指定自定义key
def Task2(**kwargs):
    file_path = kwargs['ti'].xcom_pull(task_ids='Data_Extraction_Environment', key='input_file_path')
    # ...后续处理逻辑

注意事项 #

• XCom适合传递小数据（比如字符串、小字典、数字），如果是大文件或者海量数据，别用XCom！建议把数据存在外部存储（比如S3、本地文件系统、数据库），然后只传递存储路径或标识就好。
• 一定要确保任务的依赖关系正确设置（task1 >> task2），不然task2可能在task1还没执行完就启动，会拉取不到值。

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