如何解决 Amazon MWAA 中的 "psycopg2.OperationalError: SSL connection has been closed unexpectedly" 错误？

简短描述

当您运行有向无环图 (DAG) 且任务失败时，您可能会在任务日志中收到以下错误消息：

• "psycopg2.OperationalError: SSL connection has been closed unexpectedly".
• "sqlalchemy.exc.OperationalError: (psycopg2.errors.ConnectionException) Timed-out waiting to acquire database connection".

当 Worker 过载或断开与元数据数据库的连接，或者任务尝试重用已关闭的数据库连接时，就会出现上述错误消息。

要解决此问题，请执行以下操作：

• 降低 CPU 和内存利用率。
• 检查您的 Worker 的配置。
• 使用可延迟运算符。
• 移除顶层代码。
• 不要使用 Apache Airflow 变量。
• 简化您的 DAG。
• 创建 .airflowignore 文件并将其添加到 Amazon Simple Storage Service (Amazon S3) 的主 DAG 文件夹。

解决方法

降低 CPU 和内存利用率

过载的调度器、Worker 或 Web 服务器可能无法保持活动的数据库连接。适用于 Apache Airflow 元数据的 Amazon Relational Database Service (Amazon RDS) 的默认空闲连接超时时间为 30 分钟。如果任务或进程空闲或无响应的时间超过此时间段，则数据库可能会关闭连接。有关详细信息，请参阅 IdleClientTimeout。

要检查 CPU 和内存利用率指标，请完成以下步骤：

1. 打开 Amazon CloudWatch 控制台。
2. 选择 AWS/MWAA 命名空间。
3. 在 All metrics（所有指标）选项卡中，选择 Cluster（集群）。
4. 选择 CPUUtilization，然后选择 Graph all search results（绘制所有搜索结果的图表）。
5. 选择 Graphed metrics（绘成图表的指标）选项卡以查看性能指标。
6. 重复步骤 1–3，然后选择 MemoryUtilization。
7. 选择 Graph all search results（绘制所有搜索结果的图表），然后选择 Graphed metrics（绘成图表的指标）选项卡以查看性能指标。
   **注意：**必须为 Statistic（统计数据）选择 Maximum（最大值），为 Period（时段）选择 1 minute（1 分钟）。

有关详细信息，请参阅 Amazon MWAA 的容器、队列和数据库指标。

如果您的 Worker 的 CPU 和内存利用率持续高于 90%，则 Worker 过载或任务过重。大量使用 CPU 和内存可能会导致任务状态报告、元数据数据库通信和新任务出现问题。

要降低 Worker 的 CPU 和内存利用率，请执行以下操作：

• 由于 Worker 在任务创建的初始阶段需要更多资源，因此请分散任务计划。
• 将导入范围缩小到仅限需要它们的任务，并移除不需要的全局导入。
• 使用 Apache Airflow 和 Amazon MWAA 作为主要工具。在 Amazon Elastic Kubernetes Service (Amazon EKS)、Amazon Elastic Container Service (Amazon ECS) 或 AWS Glue 等服务上安排剩余任务。
  **注意：**不要处理有关 Worker 的大量数据，因为他们的资源有限。使用专为处理大量数据而设计的服务。
• 减少任务之间传递的数据。尽可能对数据进行分块，使下游数据不会因 Amazon MWAA 工作负载过载而发生变化。
  **注意：**如果您必须在任务之间传递大量数据，请将数据存储在外部并根据需要导入。一些运算符使用 XCom。有关详细信息，请参阅 Apache Airflow 网站上的 XCom。
• 定期清理元数据数据库。
  **注意：**CPU 利用率过高或可用内存不足可能会导致查询失败或增加延迟，从而导致 SSL 异常错误。

**注意：**如果您无法减少 CPU，请使用更大的环境类，为 Worker 提供更多资源，或者进一步调整您的 Worker。

检查您的 Worker 的配置

**注意：**如果您在运行 AWS 命令行界面 (AWS CLI) 命令时收到错误，请参阅 AWS CLI 错误故障排除。此外，请确保您使用的是最新版本的 AWS CLI。

每个 Apache Airflow 版本都有 Worker 配置的默认值。但是，您可能需要根据要求自定义配置。

设置 Apache Airflow 配置

使用 Amazon MWAA 控制台、AWS CLI update-environment 命令或 UpdateEnvironment API 调用自定义以下配置：

• celery.worker_autoscale 配置控制可以同时在 Worker 上运行的最大和最小任务数。减少来自 Worker 的最大数据库连接数。然后，为具有相同 Worker 资源的较少数量的每项任务分配更多资源。
• 将 core.execute_tasks_new_python_interpreter 设置为 True，以便为每项任务创建新的 Python 解释器并隔离数据库连接。
  **注意：**只有在 "psycopg2.OperationalError" 错误再次出现时，才将配置设置为 True。
• 使用 Amazon MWAA 中的自动扩缩功能来增加最大 Worker 数。
  **注意：**同时设置 max-workers 和 celery.worker_autoscale 值以适应您的工作负载。

有关详细信息，请参阅在 Amazon MWAA 上使用 Apache Airflow 配置选项。

检查您的 Worker、调度器和 Web 服务器日志

在生产环境中，详细日志记录会使用不必要的资源并增加 CPU 利用率。最佳做法是将 Worker、调度器和 Web 服务器日志至少设置为 WARNING。在大多数情况下，不需要 INFO 日志。

使用可延迟运算符

当您的任务向外部系统或服务提交作业并等待事件发生（例如 EMR 集群创建）时，请使用可延迟运算符。可延迟运算符可以释放 Worker 空位并增加工作负载容量。有关详细信息，请参阅 Apache Airflow 网站上的可延迟运算符和触发器。

移除顶层代码

从调度器中移除使用不必要资源的顶层代码。此外，根据您的用例调整 dag_processor.min_file_process_interval 配置。有关详细信息，请参阅 Apache Airflow 网站上的 min_file_process_interval 和最佳实践。

不要使用 Apache Airflow 变量

不要在顶层 Python 代码中使用 Apache Airflow 变量。检索变量时，必须查询数据库，查询发生在每个解析循环中。

**注意：**如果必须使用变量，请使用变量缓存。有关详细信息，请参阅 Apache Airflow 网站上的 use_cache。

简化您的 DAG

简化您的 DAG，并针对 Amazon MWAA 环境的特定 Python 版本测试您的 DAG 代码。

必要时使用动态任务映射。有关详细信息，请参阅 Apache Airflow 网站上的动态任务映射。

创建 .airflowignore 文件并将其添加到 Amazon S3 中的主 DAG 文件夹

Apache Airflow 调度器将 DAG 文件夹的内容从 Amazon S3 同步到每个 Worker。调度器还会定期解析 Amazon S3 中的文件以更新用户界面。解析时，调度器运行文件并生成 DAG。

使用 .airflowignore 文件指定 Amazon S3 中 DAG 文件夹中的文件夹和文件。您必须使用正则表达式和通配符模式。在解析过程中忽略非 DAG 文件，以减少调度器上的调度器负载并提高 DAG 效率。有关详细信息，请参阅 Apache Airflow 网站上的 .airfowignore。