Airflow DAG无法运行求助：Docker环境下文件路径异常

运行Airflow DAG时触发文件不存在错误：

ERROR - [Errno 2] File b'C:/docker/docker-airflow-master/store_file/raw_store_transactions.csv' does not exist: b'C:/docker/docker-airflow-master/store_file/raw_store_transactions.csv'

本地该路径下确实存在目标文件，相关配置与代码如下：

docker-compose-LocalExecutor.yml 配置 #

version: '3.7'
services:
    postgres:
        image: postgres:9.6
        environment:
            - POSTGRES_USER=airflow
            - POSTGRES_PASSWORD=airflow
            - POSTGRES_DB=airflow
        logging:
            options:
                max-size: 10m
                max-file: "3"
                
     mysql:
        image: mysql:5.7.27
        environment:
            -MYSQL_ROOT_PASSWORD=root
        volumes:
            - ./store_file:/store_files_my_sql/
            - ./mysql.cnf:/etc/mysql/mysql.cnf

    webserver:
        image: puckel/docker-airflow:1.10.9
        restart: always
        depends_on:
            - postgres
            - mysql
        environment:
            - INSTALL_MYSQL=y
            - LOAD_EX=n
            - EXECUTOR=Local
        logging:
            options:
                max-size: 10m
                max-file: "3"
        volumes:
            - ./dags:/usr/local/airflow/dags
            -./store_file:/usr/local/airflow/store_file_airflow
            -./sql_files:/usr/local/airflow/sql_files
            # - ./plugins:/usr/local/airflow/plugins
        ports:
            - "8080:8080"
        command: webserver
        healthcheck:
            test: ["CMD-SHELL", "[ -f /usr/local/airflow/airflow-webserver.pid ]"]
            interval: 30s
            timeout: 30s
            retries: 3

Dag.py 代码 #

from airflow import DAG
from datetime import datetime, timedelta
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator

from datacleaner import data_cleaner

default_args = {
    'owner':'Airflow',
    'start_date': datetime(2022,12,14),
    'retries': 1,
    'retry_delay': timedelta(seconds=10)
}

dag=DAG('store_dag',default_args=default_args,schedule_interval='@daily',catchup=False)

t2=PythonOperator(task_id='clean_raw_csv',python_callable=data_cleaner,dag=dag)

dataCleaner.py 代码 #

def data_cleaner():

    import pandas as pd
    

    df = pd.read_csv('C:/docker/docker-airflow-master/store_file/raw_store_transactions.csv')

    def check_null():
        df.isnull().sum()
        a = df.isnull().sum()
        a.to_csv('file.csv')
        return
    check_null()

核心问题 #

Airflow运行在Docker容器内，容器无法直接访问宿主机的Windows本地路径C:/docker/...，必须通过Docker卷挂载机制实现宿主机与容器的文件共享。

具体修复步骤 #

1. 修正docker-compose的卷挂载语法
   检查webserver服务的volumes配置，当前-./store_file存在语法错误（缺少空格），修正后确保挂载路径生效：

   volumes:
       - ./dags:/usr/local/airflow/dags
       - ./store_file:/usr/local/airflow/store_file_airflow  # 补全路径前的空格
       - ./sql_files:/usr/local/airflow/sql_files
   

2. 修改dataCleaner.py的文件路径
   替换代码中的Windows本地路径为容器内的挂载路径，让Airflow能读取到共享的文件：

   def data_cleaner():
       import pandas as pd
       
       # 使用容器内的挂载路径访问文件
       df = pd.read_csv('/usr/local/airflow/store_file_airflow/raw_store_transactions.csv')
   
       def check_null():
           a = df.isnull().sum()
           # 将结果写入挂载目录，方便宿主机查看
           a.to_csv('/usr/local/airflow/store_file_airflow/file.csv')
           return
       check_null()
   

3. 重启Docker容器
   执行命令重新加载配置，确保修改生效：

   docker-compose -f docker-compose-LocalExecutor.yml down
   docker-compose -f docker-compose-LocalExecutor.yml up -d
   

额外注意事项 #

• 确认./store_file是相对于docker-compose.yml所在目录的相对路径，且宿主机该目录下确实存在raw_store_transactions.csv文件。
• 若需要在宿主机查看输出的file.csv，必须将其写入到挂载目录内，否则文件仅存在于容器内部，容器重启后会丢失。

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