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

本页的主题包含 Apache Airflow v1.10.12 Python 依赖项、自定义插件、DAG、运算符、连接、任务以及您在 Amazon MWAA环境中可能遇到的 Web 服务器问题的解决方案。

目录

• 更新 requirements.txt
  • 添加 apache-airflow-providers-amazon 会导致我的环境出现故障
• DAG 损坏
  • 我在使用 Amazon DynamoDB 运算符时收到了“DAG 损坏”的错误
  • 我收到了“DAG 损坏：没有名为 psycopg2 的模块”的错误
  • 我在使用 Slack 运算符时收到了“DAG 损坏”的错误
  • 我在安装 Google/GCP/BigQuery 时遇到了各种错误
  • 我收到了“DAG 损坏：没有名为 Cython 的模块”的错误
• 运算符
  • 我在使用 BigQuery 运算符时遇到了错误
• 连接
  • 我无法连接 Snowflake
  • 我无法连接 Secrets Manager
  • 我无法通过“<DB-identifier-name>.cluster-id.<region>.rds.amazonaws.com”连接到我的 MySQL 服务器。
• Web 服务器
  • 我正在使用 BigQueryOperator，它导致我的 Web 服务器崩溃
  • 我在访问 Web 服务器时看到 5xx 错误
  • 我看到“计划程序似乎未运行”错误
• 任务
  • 我看到我的任务卡顿或者没有完成
• CLI
  • 在 CLI 中触发 DAG 时我看到“503”错误

更新 requirements.txt

以下主题描述了您在更新 requirements.txt 时可能收到的错误。

添加 apache-airflow-providers-amazon 会导致我的环境出现故障

apache-airflow-providers-xyz 仅与 Apache Airflow v2 兼容。apache-airflow-backport-providers-xyz 与 Apache Airflow 1.10.12 兼容。

DAG 损坏

以下主题描述了您在运行 DAG 时可能收到的错误。

我在使用 Amazon DynamoDB 运算符时收到了“DAG 损坏”的错误

我们建议您完成以下步骤：

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

2. 将下列程序包引用添加到 requirements.txt 文件中。

   boto

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

我收到了“DAG 损坏：没有名为 psycopg2 的模块”的错误

我们建议您完成以下步骤：

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

2. 将以下内容添加到您 requirements.txt 文件的 Apache Airflow 版本中。例如：

   apache-airflow[postgres]==1.10.12

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

我在使用 Slack 运算符时收到了“DAG 损坏”的错误

我们建议您完成以下步骤：

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

2. 将以下程序包添加到 requirements.txt 文件并指定 Apache Airflow 版本。例如：

   apache-airflow[slack]==1.10.12

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

我在安装 Google/GCP/BigQuery 时遇到了各种错误

Amazon MWAA 使用 Amazon Linux，它需要特定版本的 Cython 和密码库。我们建议您完成以下步骤：

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

2. 将下列程序包引用添加到 requirements.txt 文件中。

   grpcio==1.27.2
   cython==0.29.21
   pandas-gbq==0.13.3
   cryptography==3.3.2
   apache-airflow-backport-providers-amazon[google]

3. 如果您不使用向后移植提供程序，则可以使用：

   grpcio==1.27.2
   cython==0.29.21
   pandas-gbq==0.13.3
   cryptography==3.3.2
   apache-airflow[gcp]==1.10.12

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

我收到了“DAG 损坏：没有名为 Cython 的模块”的错误

Amazon MWAA 使用 Amazon Linux，它需要特定版本的 Cython。我们建议您完成以下步骤：

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

2. 将下列程序包引用添加到 requirements.txt 文件中。

   cython==0.29.21

3. Cython 库有各种必需的 pip 依赖项版本。例如，使用 awswrangler==2.4.0 需要 pyarrow<3.1.0,>=2.0.0，因此 pip3 会尝试安装 pyarrow==3.0.0，这会导致“DAG 损坏”错误。我们建议明确指定可接受的最旧版本。例如，如果您在 awswrangler==2.4.0 之前指定 pyarrow==2.0.0 的最小值，则错误消失，并且 requirements.txt 安装正确。最终要求如下所示：

   cython==0.29.21
   pyarrow==2.0.0
   awswrangler==2.4.0

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

运算符

以下主题描述了您在使用运算符时可能收到的错误。

我在使用 BigQuery 运算符时遇到了错误

Amazon MWAA 不支持带有 UI 扩展的运算符。我们建议您完成以下步骤：

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

2. 解决方法是在导入问题运算符<operator name>.operator_extra_links = None后，在 DAG 中添加一行进行设置，从而覆盖扩展名。例如：

   from airflow.contrib.operators.bigquery_operator import BigQueryOperator
   BigQueryOperator.operator_extra_links = None

3. 通过将上述内容添加到插件中，您可以将此方法用于所有 DAG。有关示例，请参阅 为 Apache Airflow PythonVirtualenvOperator 创建自定义插件。

连接

以下主题描述了在使用 Apache Airflow 连接或其他 AWS 数据库时可能收到的错误。

我无法连接 Snowflake

我们建议您完成以下步骤：

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

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

   asn1crypto == 0.24.0
   snowflake-connector-python == 1.7.2

3. 将以下导入添加至 DAG：

   from airflow.contrib.hooks.snowflake_hook import SnowflakeHook
   from airflow.contrib.operators.snowflake_operator import SnowflakeOperator

确保 Apache Airflow 连接对象包含以下键值对：

1. 连接 ID：snowflake_conn

2. 连接类型：Snowflake

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

4. Schema：<my schema>

5. 登录：<my user name>

6. 密码：********

7. 端口：<port, if any>

8. 附加依赖项：

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

例如：

>>> import json
>>> from airflow.models.connection import Connection
>>> myconn = Connection(
...    conn_id='snowflake_conn',
...    conn_type='Snowflake',
...    host='YOUR_ACCOUNT.YOUR_REGION.snowflakecomputing.com',
...    schema='YOUR_SCHEMA'
...    login='YOUR_USERNAME',
...    password='YOUR_PASSWORD',
...    port='YOUR_PORT'
...    extra=json.dumps(dict(account='YOUR_ACCOUNT', warehouse='YOUR_WAREHOUSE', database='YOUR_DB_OPTION', region='YOUR_REGION')),
... )

我无法连接 Secrets Manager

我们建议您完成以下步骤：

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

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

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

我无法通过“<DB-identifier-name>.cluster-id.<region>.rds.amazonaws.com”连接到我的 MySQL 服务器。

Amazon MWAA 的安全组和 RDS 安全组需要一条入口规则来允许相互之间的流量。我们建议您完成以下步骤：

1. 修改 RDS 安全组以允许来自 Amazon MWAA 的 VPC 安全组的所有流量。

2. 修改 Amazon MWAA 的 VPC 安全组以允许来自 RDS 安全组的所有流量。

3. 再次重新运行任务，检查 CloudWatch Logs 中的 Apache Airflow 日志来验证 SQL 查询是否成功。

Web 服务器

以下主题描述了您在 Amazon MWAA 上的 Apache Airflow Web 服务器上可能收到的错误。

我正在使用 BigQueryOperator，它导致我的 Web 服务器崩溃

我们建议您完成以下步骤：

1. 诸如 BigQueryOperator 和 QuboleOperator 等 Apache Airflow 运算符包含 operator_extra_links，可能会导致 Apache Airflow Web 服务器崩溃。这些运算符试图将代码加载到 Web 服务器，但出于安全考虑，这是不允许的。我们建议通过在导入语句之后添加以下代码来修补 DAG 中的运算符：

   BigQueryOperator.operator_extra_links = None

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

我在访问 Web 服务器时看到 5xx 错误

我们建议您完成以下步骤：

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

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

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

我看到“计划程序似乎未运行”错误

如果计划程序似乎未运行，或者最后一个“心跳”是在几个小时前收到的，则 DAG 可能不会出现在 Apache Airflow 中，也不会调度新任务。

我们建议您完成以下步骤：

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

   注意

   • Aurora PostgreSQL 元数据库是 Amazon MWAA 服务架构的一部分，在 AWS 账户 中不可见。

   • 与数据库相关的错误通常是计划程序失败的症状，而不是根本原因。

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

任务

以下主题描述了在环境中执行 Apache Airflow 任务时可能收到的错误。

我看到我的任务卡顿或者没有完成

如果 Apache Airflow 任务 “卡顿” 或未完成，我们建议您执行以下步骤：

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

   

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

      

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

CLI

以下主题介绍了您在 AWS Command Line Interface 中运行 Airflow CLI 命令时可能收到的错误。

在 CLI 中触发 DAG 时我看到“503”错误

Airflow CLI 在 Apache Airflow Web 服务器上运行，该服务器的并发性有限。通常，它最多可以同时运行 4 个 CLI 命令。