Python Airflow技术问题：如何提取PythonOperator执行结果并传递给其他任务

我有一个DAG，其中调用的函数会返回一个PythonOperator。我希望获取该任务的执行结果，以便将其传递给另一个任务，请问是否可行？

以下是我的大致代码示例：

def DAG(self):

        args = {
            "owner": "airflow",
            "depends_on_past": False,
            "end_date": None,  # runs forever
            "retries": self.retries,
            "retry_delay": self.retry_delay,
            "start_date": self.start_date,
        }
        with DAG(dag_name, default_args=args, schedule_interval=dag_cron,...) as dag:
            self.parsing_tasks()
            if self.factset_cbbo_us:
                all_us_equity_mics = grab_all_mic()   # 此处返回一个PythonOperator

                generate_external_sensor_tasks(all_us_equity_mics)
            return dag

    def grab_all_mic():
        something = PythonOperator(
            task_id="blah_blah",
            python_callable=blah_blah_blah,
            op_args=("some_task", self.parsing_day_delta),
            retries=10,
            on_failure_callback=on_exhausted_retries_failure_callback,
        )
        return something

我曾找到类似问题但未得到所需答案。

完全可行，但你需要借助Airflow的XCom机制实现任务间结果传递——直接通过变量接收PythonOperator对象只能拿到任务实例，而非任务执行时的返回值。具体步骤如下：

1. 让PythonOperator的可调用函数返回目标数据 #

确保你的blah_blah_blah函数会返回需要传递的内容，示例：

def blah_blah_blah(some_task, parsing_day_delta):
    # 执行业务逻辑，生成需传递的数据集
    mic_list = ["MIC1", "MIC2", "MIC3"]
    return mic_list

2. 确保PythonOperator开启XCom推送 #

默认情况下，PythonOperator会自动将可调用函数的返回值推送到XCom，若需显式配置可添加do_xcom_push=True（默认开启，可省略）：

def grab_all_mic():
    something = PythonOperator(
        task_id="blah_blah",
        python_callable=blah_blah_blah,
        op_args=("some_task", self.parsing_day_delta),
        retries=10,
        on_failure_callback=on_exhausted_retries_failure_callback,
        do_xcom_push=True  # 可选配置，默认启用
    )
    return something

3. 在后续任务中拉取XCom数据 #

在generate_external_sensor_tasks函数中，通过任务ID从XCom拉取blah_blah任务的返回值。如果后续任务是PythonOperator，可借助ti.xcom_pull()方法实现：

def generate_external_sensor_tasks(upstream_task, dag):
    def process_mics(**context):
        # 从XCom拉取blah_blah任务的执行结果
        mic_list = context['ti'].xcom_pull(task_ids='blah_blah')
        # 用拉取到的数据集生成后续传感器任务
        for mic in mic_list:
            ExternalTaskSensor(
                task_id=f'sensor_{mic}',
                external_dag_id='your_target_external_dag',
                external_task_id=f'task_{mic}',
                dag=dag
            )
    
    # 创建处理结果的任务
    process_task = PythonOperator(
        task_id='process_mics',
        python_callable=process_mics,
        provide_context=True,  # 必须开启，才能获取包含ti的上下文
        dag=dag
    )
    # 设置任务依赖：上游任务执行完成后再运行当前任务
    upstream_task >> process_task

4. 关键注意事项 #

• 明确任务依赖：必须通过>>运算符设置任务间的依赖关系，确保后续任务在目标PythonOperator执行完成后再启动。
• XCom数据大小限制：Airflow默认限制XCom存储的数据不超过48KB，若返回的是大数据集，建议将数据存入外部存储（如数据库、对象存储），仅在XCom中存储数据的引用路径。
• 上下文参数配置：使用ti.xcom_pull()时，必须为PythonOperator设置provide_context=True（Airflow 2.x也可通过op_kwargs直接传递ti对象）。

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