MWAA自定义Operator继承Amazon Provider报错：找不到指定模块

问题背景 #

在托管式Amazon Airflow（MWAA）中开发继承自EmrContainerOperator的自定义Operator，依赖通过requirements.txt安装的Amazon Provider包，但运行时出现模块找不到的报错。

插件目录结构 #

plugins
|-- __init__.py
`-- operators
    |-- __init__.py
    `-- my_operator.py

插件定义代码（根目录__init__.py） #

from airflow.plugins_manager import AirflowPlugin

from operators.my_operator import MyOperator


class MyPlugin(AirflowPlugin):
    name = 'my_plugin'
    operators = [MyOperator]

requirements.txt内容 #

apache-airflow-providers-snowflake==3.3.0
apache-airflow-providers-amazon==2.4.0
dag-factory==0.15.0
apache-airflow-providers-slack

报错信息 #

Broken plugin: [/usr/local/airflow/plugins/operators/my_operator.py] No module named 'airflow.providers.amazon.aws.operators.emr'

解决方案 #

• 匹配MWAA版本与Provider包兼容性
  MWAA不同版本对Provider包版本有严格要求，apache-airflow-providers-amazon==2.4.0仅适配特定Airflow版本（比如MWAA 2.0.2）。若你的MWAA环境Airflow版本更高，需升级Provider包到对应兼容版本（比如3.x系列适配Airflow 2.2+）。

• 检查导入路径与文件名
  确认my_operator.py中的导入语句正确：from airflow.providers.amazon.aws.operators.emr import EmrContainerOperator；同时排查文件名拼写（报错中出现的mys_operator.py疑似笔误，需与实际文件名一致）。

• 调整插件加载顺序
  MWAA可能在Provider包安装完成前加载自定义插件，可通过延迟导入规避：
  在my_operator.py中修改初始化逻辑：

  def __init__(self, *args, **kwargs):
      from airflow.providers.amazon.aws.operators.emr import EmrContainerOperator
      super().__init__(*args, **kwargs)
  

  或在插件__init__.py中延迟导入Operator：

  from airflow.plugins_manager import AirflowPlugin
  
  class MyPlugin(AirflowPlugin):
      name = 'my_plugin'
      operators = []
  
      def __init__(self):
          from operators.my_operator import MyOperator
          self.operators = [MyOperator]
  

• 验证依赖安装状态
  查看MWAA日志控制台中requirements.txt的安装日志，确认apache-airflow-providers-amazon==2.4.0是否成功安装，排查是否存在依赖冲突（比如dag-factory与指定Provider版本不兼容）。

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