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".

上記のエラーメッセージは、ワーカーに過負荷がかかった場合、またはワーカーがメタデータデータベースへの接続を失った場合や、タスクが閉じたデータベース接続を再利用しようとした場合に発生します。

このエラーを解決するには、次の操作を行います。

• CPU とメモリの使用率を低減します。
• ワーカーの構成を確認します。
• 遅延可能な演算子を使用します。
• 最上位のコードを削除します。
• Apache Airflow 変数の使用を避けます。
• DAG を簡素化します。
• .airflowignore ファイルを作成し、Amazon Simple Storage Service (Amazon S3) のメイン DAG フォルダに追加します。

解決策

CPU とメモリの使用率を低減する

スケジューラー、ワーカー、またはウェブサーバーに過負荷がかかった場合、アクティブなデータベース接続を維持できない可能性があります。Amazon Relational Database Service (Amazon RDS) では、Apache Airflow メタデータに対するアイドル接続タイムアウトのデフォルト値は 30 分です。タスクまたはプロセスが、この期間よりも長くアイドル状態に留まったり、応答しなかったりする場合、データベースにより接続が閉じられる可能性があります。詳細については、「IdleClientTimeout」を参照してください。

CPU とメモリの使用率メトリクスを確認するには、次の手順を実行します。

1. [Amazon CloudWatch コンソール] を開きます。
2. [AWS/MWAA] 名前空間を選択します。
3. [すべてのメトリクス] タブで [クラスター] を選択します。
4. CPUUtilization を選択し、[すべての検索結果をグラフ化] を選択します。
5. [グラフ化されたメトリクス] タブを選択すると、パフォーマンスメトリクスが表示されます。
6. 手順 1 ～ 3 を繰り返し、MemoryUtilization を選択します。
7. [すべての検索結果をグラフ化] を選択し、[グラフ化されたメトリクス] を選択すると、パフォーマンスメトリクスが表示されます。
   注: [統計情報] で [最大] を選択し、[期間] では [1 分] を選択する必要があります。

詳細については、「Amazon MWAA におけるコンテナ、キュー、データベースに関するメトリクス」を参照してください。

ワーカーの CPU とメモリの使用率が常に 90% を超えている場合は、ワーカーに過負荷がかかっているか、タスクが過剰状態です。CPU とメモリの使用率が過剰な場合、タスクステータスレポート、メタデータデータベースとの通信や、新規タスクで問題が発生する可能性があります。

ワーカーにおける CPU とメモリの使用率を軽減するには、次の手順を実行します。

• ワーカーはタスク作成の初期段階で多くのリソースを必要とするため、タスクスケジュールを分散させます。
• インポートは、インポートを必要とするタスクのみに限定し、不要なグローバルインポートを削除します。
• Apache Airflow と Amazon MWAA をメインツールとして使用します。残りのタスクを、Amazon Elastic Kubernetes Service (Amazon EKS)、Amazon Elastic Container Service (Amazon ECS)、AWS Glue などのサービスでスケジュールします。
  注: ワーカーでは、リソースが限られているため、大量のデータを処理しないようにしてください。代わりに、大量のデータ処理を意図したサービスを使用してください。
• タスク間で渡されるデータを削減します。可能な場合は、Amazon MWAA ワークロードの過負荷によってダウンストリームのデータが変更されることを避けるために、データチャンクを使用します。
  注: タスク間で大量のデータを渡す必要がある場合は、データを外部に保存し、必要に応じてインポートします。一部の演算子は Xcoms を使用します。詳細については、Apache Airflow のウェブサイトで「XComs」を参照してください。
• メタデータデータベースを定期的にクリーンアップします。
  注: CPU 使用率の増加や空きメモリの減少が原因で、クエリを実行できなかったり、遅延が増加したりすると、SSL 例外エラーが発生する可能性があります。

注: CPU 使用量を削減できない場合は、大容量環境クラスを使用してワーカー用のリソースを増やすか、ワーカーで追加調整を行ってください。

ワーカーの構成を確認する

注: AWS コマンドラインインターフェイス (AWS CLI) コマンドの実行中にエラーが発生した場合は、「AWS CLI で発生したエラーのトラブルシューティング」を参照してください。また、AWS CLI の最新バージョンを使用していることを確認してください。

Apache Airflow の各バージョンには、ワーカー構成のデフォルト値が適用されます。ただし、要件によっては、構成をカスタマイズする必要が生じる場合があります。

Apache Airflow を構成する

次の構成を、Amazon MWAA コンソール、AWS CLI の update-environment コマンド、または UpdateEnvironment API コールを使用してカスタマイズします。

• celery.worker_autoscale 設定では、ワーカーで同時実行できるタスク数の最大値と最小値を制御します。ワーカーによるデータベース接続の最大数を削減します。次に、各タスクに割り当てるリソースを増やし、同じワーカーリソースにおけるタスク数を減らします。
• core.execute_tasks_new_python_interpreter を True に設定し、タスクごとに新規 Python インタープリターを作成し、データベース接続を独立させます。
  注: "psycopg2.OperationalError" エラーが再発する場合のみ、True に設定してください。
• Amazon MWAA の自動スケーリング機能を使用して最大ワーカー数を増やします。
  注: max-workers 値および celery.worker_autoscale 値の両方を、ワークロードに合わせて設定してください。

詳細については、「Amazon MWAA で Apache Airflow 設定オプションを使用する」を参照してください。

ワーカー、スケジューラー、ウェブサーバーのログを確認する

本番環境では、詳細ログ記録が不要なリソースを使用することが原因で CPU 使用率が増加します。ワーカー、スケジューラー、ウェブサーバーのログでは、基準を WARNING 以上にすることをおすすめします。ほとんどの場合、INFO ログは必要ありません。

遅延可能な演算子を使用する

タスクが外部システムまたはサービスにジョブを送信したり、イベントの発生 (例: EMR クラスターの作成) を待機したりする際には、遅延可能な演算子を使用します。延期可能な演算子を使用すると、ワーカースロットが解放され、ワークロードのキャパシティを増やすことができます。詳細については、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 を簡素化し、その DAG のコードを Amazon MWAA 環境で使用する特定の Python バージョンでテストします。

必要に応じて、動的タスクマッピングを使用します。詳細については、Apache Airflow のウェブサイトで「動的タスクマッピング」を参照してください。

.airflowignore ファイルを作成し、Amazon S3 のメイン DAG フォルダに追加する

Apache Airflow スケジューラーは、Amazon S3 内の DAG フォルダの内容を各ワーカーで同期させます。また、スケジューラーは Amazon S3 内のファイルを定期的に解析して UI を更新します。解析時に、スケジューラーはファイルを実行して DAG を生成します。

Amazon S3 内の DAG フォルダにあるフォルダとファイルを、.airflowignore ファイルで指定します。正規表現と glob パターンを使用する必要があります。解析プロセス中に、DAG 以外のファイルは考慮しないことで、スケジューラーにかかる負荷を軽減し、DAG の効率を向上できます。詳細については、Apache Airflow のウェブサイトで「.airflowignore」を参照してください。