AWS Managed Airflow 将文件夹上传到 MWAA 环境

Question

我将开始使用 AWS 管理的气流。为了让受管气流能够访问 dags，我需要将我的代码上传到 s3 存储桶中的 dags/ 目录，MWAA 会提取它。

但是，在我的代码库中，我在其他目录中有代码，例如，tasks/ 目录。问题是当我将 tasks 文件夹上传到 s3 存储桶时，mwaa 没有提取它们，并且我的 dag 出现导入错误。

AWS 文档没有为此提供任何指导。我想知道以前有没有人这样做过？还是我必须将所有代码上传到dags/ 文件夹中？

Answer 1

从 Airflow v2.x 开始，自定义代码/模块应作为常规 Python 模块导入。具体来说：

v2 中的变化

在 v2 及更高版本中，推荐的方法是将它们放在 DAG 目录中，并创建并使用 .airflowignore 文件将它们排除在被解析为 DAG 之外。

参考：https://docs.aws.amazon.com/mwaa/latest/userguide/configuring-dag-import-plugins.html#configuring-dag-plugins-changed

还是我必须将所有代码上传到 dags/ 文件夹中？

是的，您需要将代码上传到dags/ 文件夹。您还需要更新受影响的import 语句。

之前：

s3://<bucket>/dags/dag1.py
s3://<bucket>/dags/dag2.py
s3://<bucket>/tasks/task1.py
s3://<bucket>/tasks/task2.py

之后：

s3://<bucket>/dags/dag1.py
s3://<bucket>/dags/dag2.py
s3://<bucket>/dags/tasks/task1.py
s3://<bucket>/dags/tasks/task2.py
s3://<bucket>/dags/.airflowignore

Answer 2

我们使用气流版本 1.10 为 MWAA 所做的是，我们编写的任何需要在 dags 中使用的自定义代码都作为插件添加。目前，您只能根据您指定的 s3 文件夹/密钥部署 dag。我们将自定义代码放在一个单独的文件夹中，然后将其压缩并将 mwaa 配置为插件位置。

我不确定你的任务文件夹包含什么以及它有什么类型的代码，但下面是我们的代码结构示例。

src
    ------> dags (dir)
                  ----------> dag1.py
                  ----------> dag2.py
                  ----------> dag3.py
    -------> plugins(dir)
                    ----------> __init__.py
                    ----------> common (dir)
                                            -----> __init__.py
                                            -----> something.py
                    -----------> hooks (dir)
                                            ----> __init__.py
                                            ---> somehook.py
                    ------------>operators(dir)
                                           ----> __init__py
                                           ----> op1.py
                                           ----> op2.py

使用上述结构 dags 部署在 dags 文件夹中 插件文件夹完全压缩并上传到s3到mwaa中插件位置指定的位置