Airflow基础ETL任务报错求助：文件未找到与数据传递问题

• 确认文件路径有效性
  Docker部署的Airflow任务运行在容器内部，直接写Data.csv会默认在容器默认工作目录（通常为/opt/airflow）查找文件，必然找不到本地文件。解决方式：

  1. 将Data.csv放到本地Airflow挂载目录（比如./dags，对应容器内/opt/airflow/dags），代码中使用绝对路径/opt/airflow/dags/Data.csv，或通过os.path.join(os.path.dirname(__file__), 'Data.csv')获取相对当前DAG文件的路径
  2. 检查docker-compose.yml的挂载配置，确保存放文件的本地目录已被挂载到容器内，否则容器无法访问本地文件

• 检查文件权限
  容器内Airflow进程默认以airflow用户（UID/GID通常为50000）运行，若本地文件权限不足，容器内无法读取。执行chmod 644 Data.csv调整文件权限，或修改文件所属用户组匹配容器内用户

• 排查任务执行环境
  若使用CeleryExecutor/KubernetesExecutor，任务可能在独立worker容器运行，需确保所有worker能通过共享存储（如NFS、对象存储）访问文件

• 分离DAG与业务逻辑
  按计划将DAG定义与ETL业务逻辑拆分：把数据读取、处理、写入逻辑封装到独立Python模块（如etl_utils.py），放在dags目录或Airflow的PYTHONPATH路径下；DAG文件仅负责定义任务依赖、调度配置，提升代码可维护性

• 规范任务间数据传递
  Airflow任务无法直接传递DataFrame（跨进程/容器执行场景下不支持），小数据量可将DataFrame转为JSON格式通过XCom传递；大数据量需将中间结果写入共享存储（如挂载目录、数据库、对象存储），后续任务从存储读取

• 利用Airflow内置工具

  • 用FileSensor在ETL任务前检查文件是否存在，避免直接触发报错
  • 在任务函数中添加日志输出，比如打印当前工作目录print(os.getcwd())、目标文件路径，方便排查路径问题

• 添加错误处理
  在ETL函数中捕获文件异常并输出详细信息，示例代码：

  import os
  import pandas as pd
  
  def extract_data():
      file_path = os.path.join(os.path.dirname(__file__), 'Data.csv')
      print(f"读取路径: {file_path}")
      print(f"当前目录文件: {os.listdir(os.path.dirname(__file__))}")
      try:
          df = pd.read_csv(file_path)
          return df.to_json()  # 转为JSON存入XCom
      except FileNotFoundError as e:
          print(f"文件不存在: {file_path}")
          raise e
  

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