Amazon Managed Airflow (MWAA) import custom plugins

Question

I'm setting up an AWS MWAA instance and I have a problem with import custom plugins.

My local project structure looks like this:

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

I tried to match this structure in the s3 bucket:

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

However when I use the custom operator on the local project the import works like this -

from operators import customOperators

and on the MWAA it only recognize imports like this -

from plugins.operators import customOperators

Is there a way to get the MWAA recognize the import as the local (from operators)? should I upload the files in certain way to the s3?

I also tried to upload a plugins.zip file but it didn't work:

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

Answer 1

I believe the proper way is to place your custom python modules in the plugin.zip file. This file will be uploaded to MWAA and gets extracted to /usr/local/airflow/plugins/. I believe the DAGs are placed in the very same folder.

AWS has published a User Guide that gives some good explanation and examples.

Answer 2

I had the same problem and i solve it looking inside my .zip file. In my case the structure inside .zip file creates an extra folder called plugins. Review this using unzip -l plugins.zip and look the tree generated. This is my working structure:

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

Answer 3

you can import the plugin as a python-module like below

import imp

customopertaor = imp.load_source('customopertaor','/usr/local/airflow/plugins/operators/customopertaor.py')