无法从插件文件夹导入创建的包

Question

我正在使用 docker-compose 和 Dockerfile 运行 apache Airflow，并且我有以下气流文件夹结构：

dags
    example_dag
         dag_1.py
pf_utils
    __init__.py
    setup.py
    db.py --> get_primary_keys function
plugins
    _init_.py
    snowflake_operator.py --> from pf_utils.db import get_primary_keys
    

我已经从文件夹 utils 创建了一个包，以便能够按照https://airflow.apache.org/docs/apache-airflow/stable/modules_management.html#creating-a-package-in-python 中的说明在 dags 和插件代码中导入它的功能

但是，由于不存在模块，snowflake_operator 文件中的导入失败。我怎样才能让它在雪花操作符文件中工作？

如果我使用docker exec -it airflow_scheduler_1 bash 进入容器，我可以看到包pf_utils 已正确安装，原因有两个：

• 运行pip show pf_utils 向我展示：

   Name: pf-utils
   Version: 0.0.0
   Summary: UNKNOWN
   Home-page: UNKNOWN
   Author: None
   Author-email: None
   License: UNKNOWN
   Location: /home/airflow/.local/lib/python3.6/site-packages
   Requires: 
   Required-by: 
  

• 在家中 (/opt/airflow)，在 python3 控制台中运行 import pf_utils 不会出错。

但是，如果我从 /opt/airflow 移动到 /opt/airflow/plugins 并在那里启动 python3，导入将引发不存在模块的错误。在工作目录/opt/airflow/plugins 我也尝试过：

from .pf_utils.db import get_primary_keys
import .pf_utils
import ..pf_utils
from ..pf_utils import db
from airflow.pf_utils import db

以上都不行

即使添加行：

import sys
sys.path.append('/opt/airflow/pf_utils')

到snowflake_operator.py文件开头并没有解决问题。

Answer 1

如果您使用的是 Airflow 2，则不再支持从 Plugins 导入算子/传感器等

https://airflow.apache.org/docs/apache-airflow/stable/plugins.html#plugins

在 2.0 版中更改：导入操作符、传感器、添加的钩子 通过气流插件。{operators,sensors,hooks}. 是 no 更长的支持，这些扩展应该只导入为 常规的 python 模块。有关更多信息，请参阅：模块管理 并创建自定义运算符

插件曾经在 1.10 中用于此目的，但即使在那里也不需要（有时是有害的）。这不再存在 - 插件可以添加宏和自定义 UI 视图或添加“整个 Airlfow 安装行为更改”（有关何时应使用插件的示例，请参见上面的插件链接）。

在 Airflow 2（它也适用于 Airflow 1.10）中，新的操作符/传感器/钩子应仅作为常规 python 包添加：

1. 在 dags 文件夹或您在 PYTHONPATH 中添加的任何其他文件夹中（通常如果您想迭代和更改这些运算符以及 dags）。通过在 .airflowignore 中的适当排除，可以将这些运算符等排除在调度程序的解析之外：https://airflow.apache.org/docs/apache-airflow/stable/concepts/dags.html?highlight=airflowignore#loading-dags

2. 作为安装在您的环境中的常规 python 包（通常在您独立于更改 DAGS 测试和发布操作员时）。

3. 作为 Provider 包 - 如果您想对其中的一些进行逻辑分组以与特定服务交互，并且您想添加带有自定义 UI 组件的自定义连接，请添加 Operator 的额外链接 - 这将允许您注册其中一些 Hook/运营商提供新的连接和额外的链接。