Amazon MWAA部署后无法找到自定义operators及Python模块问题咨询

你遇到的ModuleNotFoundError: No module named 'operators'错误来自两个常见的配置问题：一是自定义模块目录缺少Python包必需的__init__.py标识文件，二是plugins.zip打包路径错误。

以下是两种无需使用Airflow复杂插件机制的导入方案，按需选择即可：

方案1：使用plugins.zip托管自定义模块（推荐多环境复用场景） #

1.1 正确的plugins.zip内部结构 #

打包前本地的plugins目录结构如下，所有需要作为模块导入的目录都要添加空的__init__.py文件：

plugins/
├── __init__.py  # 空文件即可
├── libs/
│   ├── __init__.py  # 空文件即可
│   ├── <my custom lib 1>.py
│   └── 其他自定义库文件
└── operators/
    ├── __init__.py  # 空文件即可
    ├── <导入自定义库1的operator>.py
    └── 其他operator文件

1.2 打包注意事项 #

打包时直接选中plugins目录下的所有文件（init.py、libs文件夹、operators文件夹）压缩为zip，不要把外层的plugins目录一起打入压缩包，否则解压后路径会变为/usr/local/airflow/plugins/plugins/导致找不到模块。

1.3 导入写法 #

• DAG文件中导入operator：from operators.<your operator file name> import <YourOperatorClass>
• Operator文件中导入自定义库：from libs.<your lib file name> import <YourUtilFunction>

方案2：直接托管到DAG目录（推荐小型项目，无需打包） #

如果你的自定义模块仅给当前环境的DAG使用，不需要跨环境复用，可以跳过plugins.zip打包步骤，直接把libs、operators目录放到S3的dags目录下，同样给每个目录添加空的__init__.py文件即可：
S3存储桶结构：

s3://{my-bucket-name}
├── dags/
│   ├── <dag 1>.py
│   ├── 其他dag文件
│   ├── libs/
│   │   ├── __init__.py
│   │   └── 自定义库文件
│   └── operators/
│       ├── __init__.py
│       └── 自定义operator文件
├── requirements.txt

导入写法和方案1完全一致，无需修改代码。

常见排查点 #

如果修改后还是报导入错误，可以临时写一个测试DAG打印Python搜索路径验证：

import sys
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def print_sys_path():
    print("Python搜索路径：", sys.path)

with DAG(
    'test_import_path',
    start_date=datetime(2024,1,1),
    schedule=None
) as dag:
    test_task = PythonOperator(
        task_id='print_sys_path',
        python_callable=print_sys_path
    )

运行后查看任务日志，确认/usr/local/airflow/plugins和/usr/local/airflow/dags都在搜索路径列表中即可。

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