如何对我的 Amazon MWAA 环境中卡滞在 Queued（排队）状态的任务进行故障排除？

简短描述

由于以下原因，Amazon MWAA 中的任务可能会卡滞在 Queued（排队）状态：

• 环境达到了最大并发任务数。
• 在您的 MWAA 环境中，气流配置选项设置不正确。
• 内存或 CPU 不足，无法在 Worker 上执行任务。

当运行任务的正常工作流程中断时，任务将卡滞在 Queued（排队）状态。Apache Airflow Worker 可能不堪重负，无法在指定的时间内做出响应。发生这种情况时，任务将保留在 Amazon Simple Queue Service (Amazon SQS) 队列中，直到 12 小时后达到默认可见性超时为止。如果您配置了重试，则 Apache Airflow 调度器会重试该任务。

解决方法

在进行故障排除之前，请确定您的环境资源是否已达到最大负载或遇到与 Worker 相关的问题。使用 Amazon CloudWatch 检查您的环境的 Worker 日志以及 CPUUtilization 和 MemoryUtilization 指标。

检查环境是否达到最大并发任务数

当 Amazon MWAA 池已满且环境向队列中添加更多任务时，您的环境将达到最大并发任务数。要解决此问题，请增加环境的 Worker 数量或更改环境类的大小。

要确定是否必须增加环境中的 Worker 数量，请完成以下步骤：

1. 打开 CloudWatch 控制台。
2. 在导航窗格中，选择 Metrics（指标），然后选择 All metrics（所有指标）。
3. 选择 Browse（浏览）选项卡，选择您的环境所在的 AWS 区域，然后搜索您的环境的名称。
4. 在 AWS Namespaces（AWS 命名空间）部分中，选择 MWAA < Queue（MWAA < 队列）。
5. 选择 QueuedTasks 和 RunningTasks。
6. 在图表中，找出活动最多的时间段，然后将这两个指标的总数相加。
   **注意：**总和是该时间段内的任务总数。
7. 确定环境的默认并发级别。
   **注意：**例如，mw1.small 环境为每个 Worker 提供五个并发任务。
8. 将任务总数除以默认的并发任务级别。
9. 将该数字减去您为环境设置的最大 Worker 数。
   **注意：**如果结果为正数，则必须添加 Worker 以完成当前的并发任务数。

要增加环境的 Worker 数量或更改环境类的大小，请完成以下步骤：

1. 打开 Amazon MWAA 控制台。
2. 选择您的环境，选择 Edit（编辑），然后选择 Next（下一步）。
3. 在 Environment Class（环境类）部分中，执行以下操作：
   增加您在步骤 9 中确定的 Maximum worker count（最大 Worker 数）。
   还需将 Minimum worker count（最小 Worker 数）设置为您的工作负载在活动最少的时段所需的值。
   **注意：**您最多只能为您的环境添加 25 个 Worker。如果您需要超过 25 个 Worker，请在 Environment class（环境类）下，选择更大的规模。
4. 如果您增加环境类大小，则还要设置工作负载所需的最大和最小 Worker 数。

如果您优化了 Worker 数，但仍不足以满足您的工作负载，请执行以下操作：

• 使用可延迟的运算符代替 Apache Airflow 传感器。有关详细信息，请参阅 Apache Airflow 网站上的可延迟运算符和触发器。
• 错开执行开始时间，并在有向无环图 (DAG) 的 schedule_interval 之间保持较小的时间间隔。以块为单位调度 DAG。
• 如果您使用自定义代码来调用和监控特定的外部函数，请将任务分成两个任务。为调用创建一项任务，并创建另一项作为可延迟的运算符来监控该函数。

检查 Airflow 配置选项是否设置不正确

要检查您的 Airflow 配置选项，请完成以下步骤：

1. 打开 MWAA 控制台。
2. 选择 Environments（环境），然后选择您的 MWAA 环境。
3. 在 Airflow Configuration options（Airflow 配置选项）部分中，检查 core.parallelism 和 celery.worker_autoscale。

如果设置了 core.parallelism，请移除任何手动设置的 core.parallelism 选项，以便 Amazon MWAA 可以动态设置配置。Amazon MWAA 通过 (maxWorkers * maxCeleryWorkers) / schedulers * 1.5 计算动态默认配置。如果您使用自动扩缩并手动设置该值，则在最大负载期间可能会出现利用率不足的问题。

将 celery.worker_autoscale 配置选项的值与默认并发级别进行比较。如果您没有修改 celery.worker_autoscale 配置选项，请将默认并发级别乘以您为环境设置的最大 Worker 数。

如果 celery.worker_autoscale 值无意中低于默认值，请使用 CloudWatch 指标监控您的 Worker 的 CPU 和内存使用情况。如果在最大负载期间资源值为 20–60%，请将 celery.worker_autoscale 值增加到更大的数字。使用较小的增量，以免过度使用 Worker 容器。

如果您未设置 celery.worker_autoscale 值或者您已保留默认值，请监控您的 Worker 的 CPU 和内存使用情况。如果您的环境的指标过高，请降低 celery.worker_autoscale 的值。如果在最大负载期间环境为 20-60%，则可以增加最大值。

检查 Worker 是否因为过度使用而失败

当 MWAA Worker 容器上的每个 Celery Worker 都有任务且处于最大负载时，Worker 可能会被过度使用并失败。

MWAA Worker 容器上的 Celery Worker 轮询当前未使用的任务。根据正在运行的任务的复杂性以及定义这些任务的代码，Worker 可能会被过度使用并可能崩溃。当 MWAA Worker 容器上的每个 Celery Worker 都有任务并且处于最大负载下时，将会发生这种情况。

要确定 Worker 是否被过度使用和失败，请完成以下步骤：

1. 打开 CloudWatch 控制台。
2. 在导航窗格中，选择 Metrics（指标），然后选择 All metrics（所有指标）。
3. 选择 Browse（浏览）选项卡，选择您的环境所在的 AWS 区域，然后搜索您的环境的名称。
4. 在 AWS Namespaces（AWS 命名空间）部分中，选择 MWAA < Queue（MWAA < 队列），然后选择 ApproximateAgeOfOldestTask。
5. 将时间范围扩大到包括 4–6 周的时间段。
   **注意：**40,000 秒或以上的峰值表明任务卡滞在 Amazon SQS 队列中，且 Worker 因过度使用而失败。此外，Celery Worker 无法将故障写入事件缓冲区，因为系统强行终止了它们。

当任务卡滞在 Amazon SQS 队列中时，您还可以使用 CloudWatch Insights 提醒您。

要创建提醒，请完成以下步骤：

1. 打开 CloudWatch 控制台。

2. 在导航窗格中，选择 Logs（日志），然后选择 Logs Insights。

3. 指定 4–6 周的时间范围。

4. 在 Selection criteria（选择标准）菜单中，选择您的 MWAA 环境的调度器日志组。

5. 在查询区中输入以下查询：

   fields _@timestamp_, _@message_, _@logStream_, _@log_  
   | filter _@message_ like /Was the task terminated externally?/  
   | sort _@timestamp_ desc  
   | limit 10000

   以下是调度器在收到先前排队的任务时发送的日志示例：

   [[34m**2024-01-17T11:30:18.936+0000**[0m] [34mscheduler_job_runner.py:[0m771 ERROR[0m - Executor reports task instance <TaskInstance: dag_name.task_name manual__202X-XX-XXTXX:XX:XX.758774+00:00 [queued]> finished (failed) although the task says it's queued. (Info: None) Was the task terminated externally?[0m

减少计算或内存密集型工作负载

**注意：**仔细考虑以下清单。并非所有因素都适用于所有使用案例。如果需要更多帮助，请联系 AWS support。

要减少环境中的计算或内存密集型工作负载，请执行以下操作：

• 确保您的 DAG 代码不包含提取、转换、加载 (ETL) 脚本、数据移动指令、AI 或 ML 管道或其他计算或内存密集型工作负载。
• 编写 DAG 代码时，请遵循 Apache Airflow 最佳实践。确保顶层代码尽量精简，并且只导入所需的内容。有关详细信息，请参阅 Apache Airflow 网站上的最佳实践。
• 优化 DAG 代码。分析任何传感器、钩子或自定义、扩展或继承操作符的内存占用量，以发现潜在的问题区域。

如果您的资源仍被过度使用，请执行以下操作：

• 将 celery.worker_autoscale 从其默认值设置减少。将 celery.worker_autoscale 值减少几位数，然后监控环境 24–48 小时。继续减少 celery.worker_autoscale 值，直到达到最佳水平。
  **注意：**当您减少 celery.worker_autoscale 值时，整个任务池会减少，并导致更多项目在更长的时间内保持在 Queue（队列）状态。为应对这种情况，您还必须增加最小 Worker 数。
• 此外，再次完成“检查环境是否达到最大并发任务数”部分中的步骤，以减少每个 Worker 的并发任务。

相关信息

Amazon MWAA 上的 Apache Airflow 性能优化

Apache Airflow 网站上的配置参考