Airflow部署时导入同目录utils模块遇ModuleNotFoundError错误

我在utils目录新增了包含字典变量的mapping.py文件，想在stage2.py中导入该变量，但部署Airflow的stage2.py DAG时抛出错误：

ModuleNotFoundError: No module named 'utils'

奇怪的是同目录下的env_config.py能正常识别。

文件夹结构 #

README.rst
LICENSE
...
dags/utils/env_config.py
dags/utils/mapping.py
dags/utils/__init__.py
dags/stage2.py
...

相关代码 #

stage2.py #

import os
import sys
from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.providers.amazon.aws.operators.glue import GlueJobOperator
from airflow.utils.task_group import TaskGroup


sys.path.insert(0, os.path.abspath(os.path.dirname(__file__)))

from utils.env_config import EnvironmentConfig  # noqa: E402
from utils.mapping import * # 此处报错


config = EnvironmentConfig(__file__)

mapping.py #

config = [
    {
        "filter_column": "SHOW_NAME",
        "horizons": [
            3,
            4,
            5,
            6
        ],
        "windows": [
            52
        ],
        "metrics": [
            "mean",
            "std",
            "min",
            "max",
            "quantile_25",
            "quantile_75",
            "trend"
        ]
    },.....
  ]

1. 检查并更新__init__.py #

utils目录下的__init__.py可能未正确导出模块，导致Airflow无法识别mapping：

• 在dags/utils/__init__.py中添加明确的导出语句：

  from .env_config import EnvironmentConfig
  from .mapping import config
  

  如果原本是空文件，添加后能让Python明确识别该目录为包。

2. 调整导入路径 #

Airflow的DAG解析环境可能对路径处理有特殊性，尝试以下两种方式：

• 相对导入：确保dags目录下存在空的__init__.py，然后替换导入语句：

  from .utils.env_config import EnvironmentConfig
  from .utils.mapping import *
  

• 绝对路径导入：修改sys.path指向dags的上级目录，再用完整路径导入：

  # 替换原sys.path语句
  sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
  
  # 导入语句改为
  from dags.utils.env_config import EnvironmentConfig
  from dags.utils.mapping import *
  

3. 清理Airflow缓存并重启服务 #

Airflow会缓存已加载的模块，新增文件后可能未刷新：

• 重启Airflow的webserver和scheduler：

  airflow webserver restart
  airflow scheduler restart
  

• 若使用Docker部署，重启对应的容器；也可以手动删除Airflow的临时缓存目录（具体路径依部署方式而定）。

4. 检查文件属性 #

• 确认mapping.py的文件名拼写、大小写和导入语句完全一致（Linux环境大小写敏感）。
• 确保Airflow运行用户拥有该文件的读取权限：

  chmod 644 dags/utils/mapping.py
  

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