Airflow EC2实例中os.getcwd()路径错误致requirements.txt无法读取

问题核心 #

你的DAG尝试通过os.getcwd()获取路径读取Airflow根目录下的requirements.txt，但调度器解析DAG时的工作目录与实际文件所在路径不匹配，导致文件找不到；同时任务运行时还存在pip权限问题。

根本原因 #

1. os.getcwd()在DAG解析阶段执行，此时的工作目录是Airflow调度器的运行目录（如/opt/***/），而非你的Airflow根目录。
2. 任务执行时，pip提示“normal site-packages is not writeable”，说明当前用户没有权限修改系统级Python包目录。

解决方案 #

方法1：使用Airflow内置环境变量AIRFLOW_HOME #

Airflow启动时会自动设置AIRFLOW_HOME环境变量，指向Airflow的根目录（即包含dags文件夹和requirements.txt的目录），直接使用这个变量是最可靠的方式。

方法2：通过DAG文件路径推导根目录 #

如果AIRFLOW_HOME未正确设置，可以通过当前DAG文件的路径向上一级推导Airflow根目录。

修改后的完整代码 #

基于AIRFLOW_HOME的实现 #

import os
from airflow import DAG
from datetime import datetime
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator

default_args = {"owner":"airflow","start_date":datetime(2022,1,1)}

# 从环境变量获取Airflow根目录，第二个参数为备用默认路径
airflow_home = os.environ.get('AIRFLOW_HOME', '/path/to/your/airflow')

def remove_modules_from_file():
    with open(f'{airflow_home}/requirements.txt', 'r+') as f:
        f.truncate(0)

with DAG(dag_id="new_modules_install",default_args=default_args,schedule_interval='@daily') as dag:
    # 使用$AIRFLOW_HOME直接引用，添加--user解决权限问题
    install_module = BashOperator(
        task_id='install_module',
        bash_command='pip install --user -r $AIRFLOW_HOME/requirements.txt'
    )
    remove_modules_from_file = PythonOperator(
        task_id='remove_modules_from_file',
        python_callable=remove_modules_from_file
    )

    install_module >> remove_modules_from_file

基于DAG文件路径的实现 #

import os
from airflow import DAG
from datetime import datetime
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator

default_args = {"owner":"airflow","start_date":datetime(2022,1,1)}

# 通过当前DAG文件路径推导Airflow根目录（当前文件在dags文件夹下，向上一级即为根目录）
dag_dir = os.path.dirname(os.path.abspath(__file__))
airflow_home = os.path.dirname(dag_dir)

def remove_modules_from_file():
    with open(f'{airflow_home}/requirements.txt', 'r+') as f:
        f.truncate(0)

with DAG(dag_id="new_modules_install",default_args=default_args,schedule_interval='@daily') as dag:
    install_module = BashOperator(
        task_id='install_module',
        bash_command=f'pip install --user -r {airflow_home}/requirements.txt'
    )
    remove_modules_from_file = PythonOperator(
        task_id='remove_modules_from_file',
        python_callable=remove_modules_from_file
    )

    install_module >> remove_modules_from_file

额外注意事项 #

• 加上--user参数是为了让pip将包安装到当前用户目录下，避开系统级目录的权限限制。
• 可在EC2实例上执行echo $AIRFLOW_HOME验证路径，确保requirements.txt确实存在于该目录中。

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