故障排除：Apache Airflow v2 中的 DAG、运算符、连接和其他问题

了解如何为 Apache Airflow 连接和变量创建密钥，请参阅 使用密钥配置 Apache Airflow 连接 Amazon Secrets Manager。

要了解如何使用 Apache Airflow 变量（test-variable）的密钥，请参阅 为 Apache Airflow 变量使用 Amazon Secrets Manager 中的密钥。

要了解如何使用密钥进行 Apache Airflow 连接（myconn），请参阅 使用 Amazon Secrets Manager 中的密钥进行 Apache Airflow 连接。

使用 GitHub 上的 aws-mwaa-local-runner 在本地测试 DAG、自定义插件和 Python 依赖项。

将以下条目添加到适合环境的 requirements.txt 中。

apache-airflow-providers-snowflake==1.3.0

将以下导入添加至 DAG：

from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator

连接 ID：snowflake_conn

连接类型：Snowflake

主机：<my account>.<my region if not us-west-2>.snowflakecomputing.com

Schema：<my schema>

登录：<my user name>

密码：********

端口：<port, if any>

附加依赖项：

{
    "account": "<my account>",
    "warehouse": "<my warehouse>",
    "database": "<my database>",
    "region": "<my region if not using us-west-2 otherwise omit this line>"
}

在 Apache Airflow UI 中查看 Amazon MWAA 提供的连接类型，请参阅 安装在 Amazon MWAA 环境中的 Apache Airflow 提供程序包。

在 CLI 中查看创建 Apache Airflow 连接的命令，请参阅 Apache Airflow CLI 命令参考。

要了解如何交替使用 Apache Airflow UI 中的连接模板来处理 Amazon MWAA 上的 Apache Airflow UI 中没有的连接类型，请参阅 连接类型概述。

检查 Apache Airflow 配置选项。验证您指定为 Apache Airflow 配置选项的键值对（例如 Amazon Secrets Manager）是否配置正确。要了解更多信息，请参阅 我无法连接 Secrets Manager。

查看 requirements.txt。验证在 requirements.txt 中列出的 Airflow “Extras”程序包和其他库是否与 Apache Airflow 版本兼容。

探索在 requirements.txt 文件中指定 Python 依赖项的方法，请参阅 在 requirements.txt 中管理 Python 依赖项。

确认 VPC 安全组允许入站访问端口 5432。需要使用此端口才能连接到环境的 Amazon Aurora PostgreSQL 元数据数据库。添加此规则后，给 Amazon MWAA 几分钟，错误就会消失。要了解更多信息，请参阅 Amazon MWAA 上的 VPC 安全。

如果计划程序未运行，则可能是由于多种因素造成的，例如依赖项安装失败或计划程序过载。在 CloudWatch Logs 中查看相应的日志组，确认 DAG、插件和要求是否正常运行。要了解更多信息，请参阅 Amazon MWAA 的监控和指标。

可能定义了大量 DAG。减少 DAG 的数量并执行环境更新（例如更改日志级别）以强制重置。

1. 无论是否启用 DAG，Airflow 都会对其进行解析。如果您使用的容量超过环境容量的 50％，则可能会开始让 Apache Airflow 计划程序不堪重负。这会导致 CloudWatch 指标中的总解析时间过长，或者在 CloudWatch Logs 中导致 DAG 处理时间过长。还有其他优化 Apache Airflow 配置的方法，这些方法不在本指南的讨论范围之内。

2. 要详细了解调整环境性能我们建议的最佳实践，请参阅 Amazon MWAA 上的 Apache Airflow 的性能调整。

队列中可能有大量任务。这通常表现为处于“无”状态的大量且不断增长的任务，或者在 CloudWatch 的排队任务和/或待处理任务中显示为大量任务。出现此错误的原因如下：

1. 要运行的任务是否多于环境的运行能力，和/或在自动扩缩有时间检测任务并部署额外的工作线程之前排队的任务数是否很多。

2. 如果要运行的任务多于环境的运行容量，我们建议减少 DAG 同时运行的任务数量，和/或增加 Apache Airflow 工作线程的最小数量。

3. 如果在自动扩缩有时间检测和部署额外工作线程之前有大量任务排队，我们建议错开任务部署和/或增加 Apache Airflow 工作线程的最小数量。

4. 您可以使用 Amazon Command Line Interface（Amazon CLI）中的 update-environment 命令来更改在环境中运行的工作线程的最小或最大数量。

   aws mwaa update-environment --name MyEnvironmentName --min-workers 2 --max-workers 10

5. 要详细了解调整环境性能我们建议的最佳实践，请参阅 Amazon MWAA 上的 Apache Airflow 的性能调整。

在执行过程中，可能会删除一些显示为任务日志但因在 Apache Airflow 中无进一步的指示而停止的任务。出现此错误的原因如下：

1. 是否存在短暂的时刻，此时 1）当前任务超出当前环境容量，然后是 2）几分钟无任务执行或排队，最后 3）新任务正在排队。

2. Amazon MWAA 自动扩缩通过添加更多工作线程来应对第一种情况。在第二种情况下，它会移除额外的工作线程。某些正在排队的任务可能会导致工作线程处于移除过程中，这些任务将在容器被删除时结束。

3. 我们建议增加环境中工作线程的最小数量。另一种选择是调整 DAG 和任务的计时，以确保这些情况不会发生。

4. 您还可以将最小工作线程数设置为等于环境中的最大工作线程数，从而有效地禁用自动扩缩。使用 Amazon Command Line Interface（Amazon CLI）中的 update-environment 命令，将最小和最大工作线程数设置为相等来禁用自动扩缩。

   aws mwaa update-environment --name MyEnvironmentName --min-workers 5 --max-workers 5

5. 要详细了解调整环境性能我们建议的最佳实践，请参阅 Amazon MWAA 上的 Apache Airflow 的性能调整。

如果任务停留在“正在运行”状态，也可以清除任务或将其标记为成功或失败。这允许环境的自动扩缩组件缩减运行在环境上的工作线程的数量。下图显示了滞留任务的示例。

1. 选择滞留任务的圆圈，然后选择清除（如图所示）。这允许 Amazon MWAA 缩减工作线程；否则，如果仍有排队的任务，Amazon MWAA 无法确定启用或禁用了哪些 DAG，也无法缩减。

   

要详细了解 Apache Airflow 任务生命周期，请参阅《Apache Airflow 参考指南》的概念。

使用 GitHub 上的 aws-mwaa-local-runner 在本地测试 DAG、自定义插件和 Python 依赖项。

创建“转换”脚本。

#!/bin/bash
cp $1 $2

（可选）macOS 和 Linux 用户可能需要运行以下命令以确保脚本可执行。

chmod 777 transform_test.sh

将脚本添加到 plugins.zip。

zip plugins.zip transform_test.sh

按照将 plugins.zip 上传到 Amazon S3 中的步骤进行操作。

按照在 Amazon MWAA 控制台上指定 plugins.zip 版本中的步骤操作。

创建以下 DAG 文件。

from airflow import DAG
from airflow.providers.amazon.aws.operators.s3_file_transform import S3FileTransformOperator
from airflow.utils.dates import days_ago
import os

DAG_ID = os.path.basename(__file__).replace(".py", "")

with DAG (dag_id=DAG_ID, schedule_interval=None, catchup=False, start_date=days_ago(1)) as dag:
    file_transform = S3FileTransformOperator(
        task_id='file_transform',
        transform_script='/usr/local/airflow/plugins/transform_test.sh',
        source_s3_key='s3://YOUR_S3_BUCKET/files/input.txt',
        dest_s3_key='s3://YOUR_S3_BUCKET/files/output.txt'
    )

按照将 DAG 代码上传到 Amazon S3 中的步骤进行操作。