如何在 IDE 中导入气流自定义运算符、插件而不会出错

Question

我不是python开发专家，这就是为什么我只列出我的步骤。

我为apache-airflow创建了一个目录~/Desktop/airflow并制作了

export AIRFLOW_HOME=~/桌面/气流

然后我创建了venv使用

python3-m venv~/桌面/气流

结果是

然后我做了

源绑定/激活

Pip3安装apache-airflow==1.10.9

气流initdb

结果是

在我的气流.cfg文件中，我检查了 dags 和插件目录。我在 $AIRFLOW_HOME/Desktop/airflow 中创建了 dags 和插件目录

我启动了气流网络服务器和调度程序，并确保一切正常。

我发现了很多创建气流插件的方法。我尝试了所有可能的方法。我们开始吧。

第一种方法是在(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)移动到$ air flow _ HOME/DESKTOP/air flow/plugins，重启airflow webserver和scheduler。

现在是时候使用我的自定义运算符创建自定义 dag了。如何正确导入插件是一个挑战。有很多可能的方法可以导入自定义运算符。我将展示我尝试过的内容。

1. 从airflow.operators导入第一操作员-已弃用
2. 从airflow.operators.first_plugin导入第一操作员
3. 从airflow.operators.first_operator导入第一操作员
4. 从first_plugin.first_operator导入第一操作员

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

从airflow.operators.first_plugin导入第一操作员

但是我确信如果我忽略导入行并将我的自定义dag放入dages文件夹，它会正常工作。（我试过了）。此外，我决定检查气流日志（在DEBUG模式下）。

当我重新启动airflow webserver时，我会看到哪些日志

我花了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

这是适用于我的文件夹结构，请确保自定义操作符位于operator文件夹中，对于传感器和 hooks也是如此。初始化。py文件应为空。

plugins
├── __init__.py
├── operators
│   ├── __init__.py
│   ├── glue_crawler_operator.py
│   └── gsheet_to_redshift_operator.py
└── sensors
    ├── __init__.py
    └── glue_crawler_sensor.py

还要检查气流中的plugins_folder。cfg文件指向插件文件夹。

例如，为了导入gsheet_to_redshift_operator运算符而不出错，我使用了以下语句：

从 operators.gsheet_to_redshift_operator 导入 GsheetToRSOperator

Answer 2

我们不需要创建文件夹< code >插件和添加< code >自定义操作符。有了Airflow 2.0，我们只需要在项目中创建一个文件夹，并将其导入dag文件中。

airflow_project
+--custom_operator
   +--first_operator.py
+--dags
   +--dag_1.py

在dag_1.py中，您可以将自定义运算符/钩子导入为：

from custom_operator.first_operator import FirstOperator

Answer 3

我绕过了插件步骤，以一种纯python式的方式完成了它。这似乎也是文档对最新版本（1.10.12）的说法。

这是我的文件夹结构：

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

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

from custom_operator.hello_operator import HelloOperator

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