如何使用Airflow本地写入文件？执行DAG遇目录不存在错误

错误原因 #

报错的核心原因是代码中指定的目标目录/Users/paul/airflow/output不存在，Python的open()函数以'w'模式打开文件时，只会自动创建文件，但不会自动创建不存在的父目录。

解决方案 #

1. 在代码中自动创建目标目录 #

修改你的write_simple_file函数，添加目录检查与创建逻辑，使用os.makedirs()递归创建所需目录（如果不存在）：

from datetime import datetime, timedelta
import os  # 新增导入os模块
from airflow import DAG
from airflow.operators.python import PythonOperator

def write_simple_file(**kwargs):
    file_path = '/Users/paul/airflow/output/hello_airflow.txt'
    # 获取文件所在的父目录路径
    dir_path = os.path.dirname(file_path)
    # 递归创建目录，exist_ok=True避免目录已存在时报错
    os.makedirs(dir_path, exist_ok=True)
    with open(file_path, 'w') as file:
        file.write("Hello from Airflow DAG!\n")
    print(f"File written to {file_path}")

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'write_file_dag',
    default_args=default_args,
    description='DAG pour écrire un fichier simple',
    schedule_interval=None,  
    start_date=datetime(2024, 5, 7),
    catchup=False,
)

write_file_task = PythonOperator(
    task_id='write_file_task',
    python_callable=write_simple_file,
    dag=dag,
)

write_file_task

2. 确认目录权限 #

如果创建目录后仍然报错，需要检查Airflow运行用户对目标目录的读写权限：

• 查看Airflow的运行用户：执行ps aux | grep airflow获取进程所属用户
• 给目标目录添加对应权限：比如执行sudo chown -R airflow:airflow /Users/paul/airflow/output（替换为实际的Airflow用户）
• 或者选择Airflow默认有权限的目录，比如Airflow HOME目录下的output子目录，可通过os.environ.get('AIRFLOW_HOME')获取默认路径。

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