Airflow DAG跨任务是否保留临时文件？求助解决文件找不到报错

一、跨任务是否保留临时文件？ #

不会。Airflow的每个任务默认运行在独立的执行环境中：

• 若使用CeleryExecutor等分布式执行器，不同任务可能跑在不同Worker节点，本地文件完全不互通；
• 即使是SequentialExecutor，每个任务的工作目录也是临时生成的，任务结束后本地文件无法被后续任务访问。

所以t1生成的download.csv属于t1的本地临时文件，t2根本找不到。

二、解决方法 #

1. 用共享存储存储文件（推荐） #

让所有任务挂载同一个可访问的共享存储（比如NFS、S3、HDFS，或者Airflow部署节点的固定目录），指定统一的文件路径：

• 修改api_download_lineitemstatus.py：将文件写入共享路径，比如open('/airflow_shared/download.csv', 'w')
• 修改lineinsert.py：从同一个共享路径读取文件，比如open('/airflow_shared/download.csv', 'r')
• 确保所有Airflow Worker对该路径有读写权限。

2. 用XCom传递小数据（适合文件内容不大的场景） #

如果download.csv内容较小，直接通过Airflow的XCom机制传递数据，不需要依赖文件：

• 在api_download_lineitemstatus.py中，生成文件后读取内容，用XCom推送：

  from airflow.utils.xcom import push
  from airflow.models import TaskInstance
  from airflow.utils.session import create_session
  import os
  from datetime import datetime
  
  execution_date = datetime.fromisoformat(os.environ['AIRFLOW_CTX_EXECUTION_DATE'])
  with open('download.csv', 'r') as f:
      data = f.read()
  with create_session() as session:
      ti = TaskInstance(task_id='api_download', execution_date=execution_date)
      push(ti=ti, key='csv_data', value=data)
  

• 在lineinsert.py中，从XCom拉取数据直接处理：

  from airflow.utils.xcom import pull
  from airflow.models import TaskInstance
  from airflow.utils.session import create_session
  import os
  from datetime import datetime
  
  execution_date = datetime.fromisoformat(os.environ['AIRFLOW_CTX_EXECUTION_DATE'])
  with create_session() as session:
      ti = TaskInstance(task_id='insert_into_DB', execution_date=execution_date)
      data = pull(ti=ti, key='csv_data', task_ids='api_download')
  # 处理data并插入数据库
  

3. 合并两个任务（简单场景适用） #

如果两个任务逻辑不复杂，直接合并成一个任务，在同一个执行环境中完成下载和插入：

t_combined = BashOperator(
    task_id="download_and_insert",
    bash_command='api_download_lineitemstatus.py && python lineinsert.py',
)

或者将两个脚本的逻辑整合到一个Python函数中，用PythonOperator执行。

4. 本地固定路径（不推荐，仅单节点环境可用） #

如果是单节点部署（比如用SequentialExecutor），可以指定一个本地固定路径（比如/tmp/airflow_temp/download.csv），让t1写入、t2读取。但这种方式无法适配分布式场景，且多DAG实例同时运行时会出现文件覆盖问题，不建议用于生产环境。

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