Amazon Managed Airflow (MWAA)导入自定义插件

Question

我正在设置AWS MWAA实例，导入自定义插件时遇到问题。

我的本地项目结构如下所示：

airflow-project
├── dags 
│   └── dag1.py  
└── plugins
    ├── __init__.py
    └── operators
        ├── __init__.py
        └── customopertaor.py

我尝试在s3存储桶中匹配此结构：

s3://{my-bucket-name}
└── DAGS
    ├── dags 
    │   └── dag1.py  
    └── plugins
        ├── __init__.py
        └── operators
            ├── __init__.py
            └── customopertaor.py

但是，当我在本地项目上使用自定义操作符时，导入的工作方式如下所示-

from operators import customOperators

在MWAA上，它只识别像这样的进口-

from plugins.operators import customOperators

有没有办法让MWAA将导入识别为本地(从运算符)？我应该以某种方式将文件上传到s3吗？

我还试图上传一个plugins.zip文件，但没有成功：

s3://{my-bucket-name}
├── DAGS
│   └── dags 
│       └── dag1.py  
└── plugins.zip

Answer 1

我认为正确的方法是将您的自定义python模块放在plugin.zip文件中。此文件将上载到MWAA并解压到/usr/local/airflow/plugins/。我相信DAG放在同一个文件夹中。

亚马逊网络服务已经发布了一个User Guide，给出了一些很好的解释和例子。

Answer 2

我也有同样的问题，我通过查看我的.zip文件解决了这个问题。在我的例子中，.zip文件内部的结构创建了一个名为plugins的额外文件夹。使用unzip -l plugins.zip查看并查看生成的树。这是我的工作结构：

Archive:  plugins.zip
Length      Date    Time    Name
    0  10-18-2021 11:39   hooks/
  125  10-18-2021 11:40   hooks/my_airflow_hook.py
    0  10-18-2021 11:40   sensors/
  359  10-18-2021 11:40   sensors/my_airflow_sensor.py
  395  10-18-2021 13:28   my_airflow_plugin.py
    0  10-18-2021 11:42   operators/
  437  10-18-2021 11:42   operators/hello_operator.py
  480  10-18-2021 11:42   operators/my_airflow_operator.py