问题标签 [mwaa]

我在解析 DAG 时遇到错误：

Broken DAG: [/usr/local/airflow/dags/test.py] No module named 'airflow.providers'

我将 apache-airflow-providers-databricks 添加到 requirements.txt，并从日志中看到：

Successfully installed apache-airflow-2.0.1 apache-airflow-providers-databricks-1.0.1 apache-airflow-providers-ftp-1.0.1 apache-airflow-providers-http-1.1.1 apache-airflow-providers-imap-1.0.1 apache-airflow-providers-sqlite-1.0.2 apispec-3.3.2 attrs-20.3.0 cattrs-1.3.0 clickclick-20.10.2 commonmark-0.9.1 connexion-2.7.0 flask-appbuilder-3.1.1 flask-caching-1.10.0 gunicorn-19.10.0 importlib-resources-1.5.0 inflection-0.5.1 isodate-0.6.0 marshmallow-3.10.0 marshmallow-oneofschema-2.1.0 openapi-schema-validator-0.1.4 openapi-spec-validator-0.3.0 pendulum-2.1.2 python-daemon-2.3.0 rich-9.2.0 sqlalchemy-jsonfield-1.0.0 swagger-ui-bundle-0.0.8 tenacity-6.2.0 termcolor-1.1.0 werkzeug-1.0.1

但是调度程序似乎卡住了：

The scheduler does not appear to be running. Last heartbeat was received 19 hours ago.

我怎样才能重新启动它？

AWS MWAA（Apache Airflow 托管工作流）是 AWS 提供的相对较新的服务。配置 MWAA 环境时，可以提供自定义requirements.txt文件，用于在该环境中安装额外的 Python 包。

在我工作的公司中，我们将AWS CodeArtifact用于自定义 PyPi 包存储库，我们在其中上传私有 Python 包。我们想在 Airflow DAG 中使用其中的一些。这就是为什么我想知道是否可以以某种方式配置 MWAA 环境以使用 CodeArtifact 中的 PyPi 存储库？

或者有什么方法可以在 MWAA 环境中安装自定义 Python 包（不在公共 PyPi 中）？

我需要将文件从 s3 下载到 aiflow tmp\test\ 文件夹以处理 python 脚本并将文件加载回 s3。

Step1：创建文件夹
Step2：使用 bash 脚本下载文件 1
Step3：使用 bash 脚本下载文件 2
Step4：使用 bash 脚本下载文件 3
Step5：使用 bash 脚本下载文件 4
Step6：使用 bash 脚本下载文件 5
Step7：执行 python 脚本
第 8 步：将 python 输出加载到 s3 第 2 步到第 6 步（下载）在 DAG 中并行运行。

我创建了 DAG，它第一次运行良好。当我第二次下载很少的文件并且由于找不到模板/tmp/test/而导致很少的下载步骤（步骤 3 和步骤 5）异常时。

我正在尝试按照此处所述暂停 DAG，但即使 DAG 存在，也会出现“找不到 DAG id testDag”。

错误消息说我的第 3 方模块“未找到”，即使它在我触发 DAG 时工作。

我在文档上看到了这个：

注意 如果 DAG 使用依赖于通过 requirements.txt 安装的包的插件，则任何解析 DAG 的命令（例如 list_dags、backfill）都将失败。

Dag 代码（testDag）：

气流API：

我可以使用的任何解决方法？谢谢

我需要在代码中更改哪些内容以添加标签和气流配置选项？

如果我使用tags={'Environment': 'Example'}我会收到以下错误

我最近开始研究 Amazon MWAA，并且能够设置环境并上传 DAG。当我尝试运行该作业时，它会失败而不会创建任何日志。 失败任务的快照

如果有人可以看一下，将不胜感激:)

下面是 DAG 代码：

我正在尝试运行相当于backfill -t TASKNAME DAGNAME

我从这里复制了脚本，但在我看来，运行 cli 的环境没有加载与我相同的包requirements.txt（Airflow 可以解析 DAG）

结果是这个

我已经手动配置了一个 Glue Crawler，现在正尝试通过 Airflow（在 AWS 中）运行它。

根据此处的文档，与 Glue 环境中的其他任务相比，似乎有很多方法可以处理此目标。但是，我在处理这个看似简单的场景时遇到了问题。

以下代码定义了 Glue[Crawler]+Airflow 的基本设置。假设在它之前和之后定义了一些其他的工作任务，这里不包括。

现在，这是一个示例流程：

鉴于这一切，Airflow Webserver UI 上会出现以下错误：

我明白了：你为什么不使用start_crawler方法以外的东西......？公平点，但我不知道还能用什么。我只想在一些上游任务成功完成但无法启动爬虫。

我应该如何解决这个问题？

我正在尝试在 AWS MWAA 上构建 DAG，此 DAG 会将数据从 Postgres (RDS) 导出到 S3，但是一旦 MWAA 尝试解析我的任务中的所有查询，它就会出现问题，总共它将导出 385 个表，但 DAG 卡在运行模式下，无法启动我的任务。

基本上，这个过程将：

1. 加载表架构
2. 重命名某些列
3. 将数据导出到 S3

功能

任务

气流 list_dags 输出

观察

如果我只允许在任务中加载一个表，它工作得很好，但如果所有表都可以加载，则失败。如果从指向 RDS 的 docker 执行 Airflow，此行为是相同的

气流 list_dags 的屏幕截图：

我们正在使用 AWS MWAA。我们以编程方式将 DAG.py 文件添加到 S3 存储桶中。然后它们会显示在 UI 中。但是，它们是“关闭”的，您必须单击“打开”按钮才能启动它们。

编辑：此外，我们有时可能想要将打开到关闭的 DAG（以编程方式）

我希望以编程方式执行此操作，但我无法弄清楚。

API 似乎没有： https ://docs.aws.amazon.com/mwaa/latest/userguide/mwaa-actions-resources.html

Boto 似乎没有： https ://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/mwaa.html

是否可以通过 API 从 OFF/ON ON/OFF 操作 DAG 状态？