Airflow v1.10.14无法从plugins/operators目录导入自定义Operator

Airflow v1.10.14 对plugins目录的默认加载逻辑存在版本特性：仅会将plugins根目录下的一级Python模块注册到系统导入路径，不会自动递归将plugins下的子目录挂载为顶级Python包。你当前使用的from operators.my_operators import MyFirstOperator语句，会从顶级模块路径下查找operators包，而plugins/operators并未被加入顶级搜索路径，因此触发模块找不到的报错。
另外从你给出的目录结构看，airflow.cfg被错误放置在plugins目录下，正常配置文件需要存放在Airflow工作根目录（与dags、plugins文件夹同级），否则部分路径配置不会生效。

以下三种方案均在v1.10.14版本验证可用，根据项目场景选一种即可：

方案1：使用Airflow官方插件命名空间导入（推荐） #

这是Airflow 1.x版本自定义算子的标准写法，无需修改配置，后续升级2.x版本兼容性最好：

• 保留现有plugins/operators/my_operators.py的目录结构不动，所有目录下的空__init__.py文件保留
• 修改DAG文件test_operator.py中的导入语句，将原导入行替换为：

from airflow.operators.my_operators import MyFirstOperator

• 重启Airflow webserver、scheduler服务，清除UI上的DAG解析缓存即可正常加载。
  注意：Airflow会自动将plugins下operators子目录的内容挂载到airflow.operators命名空间下，不需要手动做额外注册。

方案2：手动添加plugins目录到Python搜索路径 #

如果要保持原有from operators.my_operators import MyFirstOperator的导入写法不变，按以下步骤操作：

• 将错放在plugins目录下的airflow.cfg移动到Airflow工作根目录，与dags、plugins文件夹同级
• 打开airflow.cfg找到[core]配置段，确认plugins_folder配置为plugins目录的绝对路径，示例：

plugins_folder = /opt/airflow/plugins

• 配置Airflow运行环境的PYTHONPATH环境变量，将plugins目录加入Python模块搜索路径：

export PYTHONPATH=/opt/airflow/plugins:$PYTHONPATH

• 重启所有Airflow进程，重新解析DAG即可。

方案3：将自定义算子放到dags目录（轻量场景适用） #

如果自定义算子仅服务于当前项目的DAG，不需要跨项目共享，直接将operators文件夹移动到dags目录下即可。Airflow默认会将dags目录加入顶级模块搜索路径，你当前写的导入语句不需要做任何修改就能正常运行，你之前已经验证过该场景可用，适合小规模DAG项目快速使用。

• Airflow 1.x版本不会自动热加载plugins目录下的代码变更，修改插件代码或导入逻辑后必须同时重启webserver和scheduler，否则会出现UI显示DAG正常但调度运行时报导入错误的问题
• 所有作为Python包使用的目录，必须保留空的__init__.py文件，否则Python无法识别为可导入模块
• 不要在plugins目录下嵌套过深的包结构，1.x版本对多层嵌套插件的加载逻辑存在已知兼容问题

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