python - 如何在 IDE 中导入 Airflow 自定义操作符、插件而不会出错

Question

我不是专业的 Python 开发人员，这就是为什么我只会 概述我的步骤 .
设置部分
我创建了一个目录 ~/桌面/Airflow 用于 apache-airflow 并制作

export AIRFLOW_HOME=~/Desktop/airflow

然后我创建了 venv 使用

python3 -m venv ~/Desktop/airflow

结果是

然后我做了

source bin/activate

pip3 install apache-airflow==1.10.9

airflow initdb

结果是

在我的 Airflow .cfg 文件我检查了 dags 和 plugins 目录。我在 中创建了 dags 和 plugins 目录$AIRFLOW_HOME/桌面/Airflow
我开始了 Airflow 网络服务器 和 调度程序并确保一切正常。
自定义插件部分
我找到了很多方法来创建 Airflow 插件。我尝试了所有可能的方法。开始吧。
第一个是在( first_plugin )一个项目中创建一个插件文件夹，然后创建一个python文件( first_operator.py )

import logging

from airflow.operators import BaseOperator
from airflow.utils.decorators import apply_defaults
from airflow.plugins_manager import AirflowPlugin

log = logging.getLogger(__name__)


class FirstOperator(BaseOperator):

    @apply_defaults
    def __init__(self, *args, **kwargs):
        super(FirstOperator, self).__init__(*args, **kwargs)

    def execute(self, context):
        log.info("Hello World!")


class FirstOperatorPlugin(AirflowPlugin):
    name = "first_plugin"
    operators = [FirstOperator]

看起来像

然后我只需将我的插件文件夹( first_plugin )移动到 $AIRFLOW_HOME/DESKTOP/airflow/plugins 并重新启动 Airflow 网络服务器和调度程序。
现在是使用我的自定义运算符创建自定义 dag 的时候了。如何正确导入您的插件是一个挑战。有很多方法可以导入自定义运算符。我会展示我的尝试。

从airflow.operators 导入FirstOperator - 已弃用

来自 Airflow 。运营商。 first_plugin 导入 FirstOperator

来自 Airflow 。运营商。 first_operator 导入 FirstOperator

从 first_plugin.first_operator 导入 FirstOperator

在 Pycharm IDE 中导入时，这些方法都没有帮助我。例如，

from airflow.operators.first_plugin import FirstOperator

但我确定如果我忽略导入行并将我的自定义 dag 放入 dags 文件夹，它会正常工作。 (我试过)。此外，我决定检查 Airflow 日志(在 Debug模式下)。
重新启动 Airflow 网络服务器时看到的日志

我花了2天，我仍然没有任何解决方案。可能你们告诉我尝试其他方法。我试过。
https://www.astronomer.io/guides/airflow-importing-custom-hooks-operators/https://www.astronomer.io/guides/airflow-importing-custom-hooks-operators/
https://pybit.es/introduction-airflow.html
所有这些都是有效的方法，但没有一个能解决我的 IDE 导入问题。

Answer 1

我绕过了插件步骤，并以纯 Python 风格的方式完成了。似乎这也是最新版本(1.10.12)的文档所说的。
这是我的文件夹结构:

AIRFLOW_HOME
├── dags
│   ├── __init__.py
│   └── my_dag.py
└── utils
    ├── __init__.py
    └── custom_operator.py

确保 PYTHONPATH包含 AIRFLOW_HOME小路。然后你可以像这样导入它:

from custom_operator.hello_operator import HelloOperator

在此处阅读更多信息: https://airflow.apache.org/docs/stable/howto/custom-operator.html#creating-a-custom-operator