トラブルシューティング: DAGs、オペレータ、接続、その他の問題

Apache Airflow 接続と変数のシークレットキーを作成する方法を AWS Secrets Manager シークレットを使用した Apache Airflow 接続の設定 で学習できます。

test-variable で ApacheAirflow 変数 (Apache Airflow 変数 AWS Secrets Manager の でのシークレットキーの使用) のシークレットキーを使用する方法を学習します。

Apache Airflow 接続 AWS Secrets Manager に でシークレットキーを使用する で ApacheAirflow 接続 (myconn) のシークレットキーを使用する方法を学習します。

GitHub DAGs、カスタムプラグイン、Python 依存関係をローカルでテストします。 aws-mwaa-docker-images

ご使用の環境に適した requirements.txt に次のエントリを追加します。

apache-airflow-providers-snowflake==1.3.0

以下のインポートを DAG に追加する：

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

接続 ID: snowflake_conn

コーンタイプ: スノーフレーク

ホスト： <my account> <my region if not us-west-2>.snowflakecomputing.com

スキーマ: <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 エアフロープロバイダーパッケージ。

コマンドにアクセスして、 の CLI で Apache Airflow 接続を作成しますApache Airflow CLI コマンドリファレンス。

接続タイプの概要 Amazon MWAA 上の Apache Airflow UI では使用できない接続タイプのために、Apache Airflow UI で接続テンプレートを交換して使用する方法を学習します。

Apache Airflow 構成オプションを確認してください。などの Apache Airflow 設定オプションとして指定したキーと値のペアが正しく設定 AWS Secrets Managerされていることを確認します。詳細については、「Secrets Manager に接続できません。」を参照してください。

requirements.txt をチェックしてください。Apache Airflow バージョンと互換性のある、requirements.txt にリストされている Airflow の「extras」パッケージおよびその他のライブラリを確認してください。

requirements.txt ファイルで Python 依存関係を指定する方法については、「」を参照してくださいrequirements.txt での Python 依存関係の管理。

VPC セキュリティグループがポート 5432 へのインバウンドアクセスを許可していることを確認します。これは、ご使用の環境の Amazon Aurora PostgreSQL メタデータデータベースに接続するために必要のポートです。このルールを追加したら、Amazon MWAA に数分与えると、エラーが消える可能性があります。詳細については、「Amazon MWAA の VPC のセキュリティ」を参照してください。

スケジューラーが動作していない場合は、依存関係のインストールの失敗 や スケジューラーの過負荷 など、さまざまな要因が原因である可能性があります。CloudWatch Logs DAGs、プラグイン、要件が正しく動作していることを確認します。詳細については、「Amazon Managed Workflows for Apache Airflow のモニタリングとメトリクス」を参照してください。

多数の DAGs が定義されている可能性があります。DAG の数を減らし、環境の更新 (ログレベルの変更など) を実行して強制的にリセットしてください。

1. Airflow は DAG が有効かどうかに関係なく解析します。環境の容量の 50% 以上を使用している場合は、Apache Airflow スケジューラの負担が増大する可能性があります。これにより、CloudWatch メトリクスの合計解析時間が長くなったり、CloudWatCloudWatch Logs の DAG 処理時間が長くることがあります。Apache Airflow の設定を最適化する方法は他にもありますが、このガイドの対象範囲には含まれていません。

2. 環境のパフォーマンスを調整するために推奨されるベストプラクティスの詳細については、「」を参照してくださいAmazon MWAA での Apache Airflow のパフォーマンス調整。

キューには多数のタスクが存在する可能性があります。これは多くの場合、 None状態のタスクの数が大きい、増加している、または Tasks Pending CloudWatch Queued Tasksのタスクの数が多いと表示されます。これは以下のような原因で起こりうる：

1. 実行するタスクが環境の容量よりも多い場合、および/または Auto Scaling がタスクを検出して追加のワーカーをデプロイする前にキューに入れられたタスクが多数ある場合。

2. 実行するタスクが環境の実行容量よりも多い場合は、DAGs が同時に実行するタスクの数を減らすか、Apache Airflow ワーカーの最小数を増やすことをお勧めします。

3. Auto Scaling が追加のワーカーを検出してデプロイする前にキューに入れられたタスクが多数ある場合は、タスクのデプロイをずらすか、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 のパフォーマンス調整。

タスクが「実行中」状態のままになっている場合は、タスクをクリアしたり、成功または失敗のマークを付けることもできます。これにより、環境の自動スケーリングコンポーネントは、環境で実行されているワーカー数をスケールダウンできます。次の図は、ストランドタスクの例を示しています。

1. 取り残されたタスクの円を選択し、クリアを選択します（図を参照）。これにより Amazon MWAA はワーカーをスケールダウンできます。そうしない場合、Amazon MWAA はどの DAG が有効か無効かを判断できず、キューにタスクが残っている場合はスケールダウンできません。

   

Apache Airflow タスクライフサイクルの詳細については、Apache Airflow リファレンスガイドの 「概念」 を参照してください。

GitHub DAGs、カスタムプラグイン、Python 依存関係をローカルでテストします。 aws-mwaa-docker-images

「変換」スクリプトを作成します。

#!/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://amzn-s3-demo-bucket/files/input.txt',
						dest_s3_key='s3://amzn-s3-demo-bucket/files/output.txt'
						)

「Amazon S3 への DAG コードのアップロード」 のステップに従ってください。