Airflow任务间传递自定义对象类型数据及模块导入问题

一、任务间传递数据的方法 #

Airflow的任务是独立运行的（可能在不同Worker进程/容器），没法直接传递内存中的自定义对象，推荐用以下几种方式：

1. 用Airflow自带的XCom（适合小数据量） #

XCom是Airflow内置的任务间数据传递机制，能存小批量的序列化数据。注意：

• 第二个系统的自定义对象默认无法直接序列化，建议先把数据转成dict/list这类原生可序列化结构，传递后再重新实例化对象。
• 示例代码：

  # 第一个任务：读取第一个系统数据并存入XCom
  def fetch_first_system_data(**context):
      import requests
      first_data = requests.get("你的第一个系统API地址").json()
      # 存入XCom，指定key方便后续获取
      context["ti"].xcom_push(key="first_system_data", value=first_data)
  
  # 第二个任务：读取第二个系统数据，拉取XCom数据对比更新
  def sync_second_system(**context):
      from second_system_module import fetch_objects
      # 从XCom拉取第一个系统的数据
      first_data = context["ti"].xcom_pull(key="first_system_data")
      # 用第二个系统模块获取自定义对象
      second_objects = fetch_objects()
      # 对比并更新
      for obj in second_objects:
          if obj.id in first_data:
              obj.field = first_data[obj.id]["target_value"]
              obj.save()  # 模块自动处理API请求
  

• 若必须传递自定义对象，需在Airflow配置airflow.cfg中开启enable_xcom_pickling = True，但这种方式存在安全风险（pickle反序列化可能被利用），不推荐。

2. 用外部存储（适合大数据量） #

如果数据量超过XCom默认限制（约48KB），用以下方式：

• 将数据写入CSV/JSON文件，存储到共享磁盘、S3或其他分布式存储，后续任务直接读取文件。
• 存入数据库（比如PostgreSQL、MySQL），第一个任务写库，第二个任务读库。

二、第二个系统Python模块的导入位置 #

根据你的部署场景，选以下一种方式：

1. 放到Airflow的DAG目录 #

把模块文件（比如second_system_module.py）直接放在Airflow的dags文件夹下，这样DAG文件就能直接import second_system_module。

• 优点：简单直接，无需额外配置。
• 注意：如果模块有第三方依赖（比如requests），要确保Airflow的Python环境已经安装这些依赖。

2. 安装到Airflow的Python环境 #

如果你的模块是一个可安装的Python包（有setup.py或pyproject.toml），直接在Airflow运行的Python环境中执行pip install .（本地包）或pip install 包名（PyPI包），这样所有DAG都能导入该模块。

• 适合模块需要被多个DAG复用的场景。

3. 添加到PYTHONPATH（不推荐，仅临时场景用） #

如果模块在其他目录，可在DAG文件开头添加：

import sys
sys.path.append("/path/to/your/module/directory")
import second_system_module

• 缺点：不够优雅，部署到多Worker环境时要确保所有Worker都能访问该目录。

容器化部署注意事项 #

如果用Docker/K8s部署Airflow，要把模块打包到Airflow镜像中，或者将模块目录挂载到容器内的dags目录或PYTHONPATH路径下。

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