Amazon MWAA (Apache Airflow)

(Amazon Managed Workflows for Apache Airflow) は、オープンソースのワークフロー管理ツールを、AWS がマネージドで動かすサービスです。Airflow のスケジューラー、ワーカー、Web サーバー、メタデータベースの構築とパッチ適用を AWS に任せ、利用者は Python で書いたワークフロー () をに置くだけで動かせます。

DEA では、DAG の置き場所と依存パッケージの入れ方、環境クラスとワーカーの自動スケーリング、タスクが失敗したときの調べ方 ( Logs と Airflow UI) が問われます。

Point

MWAA = Managed Workflows for Apache Airflow の略です。試験ガイドでも「Amazon Managed Workflows for Apache Airflow (Amazon MWAA)」と書かれます。
DAG = Directed Acyclic Graph (有向非巡回グラフ) です。タスクを矢印でつなぎ、輪にならない形で順序を表します。
Operator (オペレーター) は「1つのタスクで何をするか」の部品です。Sensor (センサー) は「条件が満たされるまで待つ」オペレーターです。
和訳
- Airflow: 空気の流れ。ここでは「仕事の流れ」を表す製品名です。
- Weblio英和辞書「airflow」

概要

の環境は、AWS が管理するの各コンポーネントと、利用者のとから成ります。

構成要素	役割
S3 バケット	のファイル (`dags/` フォルダ)、`requirements.txt`、`plugins.zip`、起動スクリプトを置きます。
スケジューラー	DAG を読み込み、実行の予定を立ててタスクをキューに入れます。
ワーカー	キューからタスクを取り出して実行します。負荷に応じて台数が増減します。
Web サーバー	Airflow UI を提供します。アクセスはで制御します。
メタデータベース	DAG の実行状態を保存します。AWS が管理します。
実行ロール	ワーカーが AWS のサービスを呼ぶときの IAM ロールです。

試験では、「既存の Airflow の DAG を、インフラを管理せずに AWS で動かしたい」「Python でワークフローを書き、AWS 外のシステムも扱いたい」という要件で MWAA を選びます。

主要技術

①DAG、オペレーター、センサー ⚠️ 頻出

は Python のファイルで、タスクとその依存関係を定義します。各タスクはオペレーターで作ります。

部品	例	データパイプラインでの使い方
AWS のオペレーター	GlueJobOperator、EmrAddStepsOperator、AthenaOperator、RedshiftDataOperator、S3CopyObjectOperator など	AWS のサービスを起動し、完了を待ちます。
センサー	S3KeySensor、GlueJobSensor など	にファイルが届くまで、ジョブが終わるまで待ちます。
汎用オペレーター	PythonOperator、BashOperator	小さな Python 処理やコマンドを実行します。
依存関係	`extract >> transform >> load`	前のタスクが成功してから次を始めます。

AWS のオペレーターとセンサーは、の Amazon プロバイダーパッケージに含まれます。Airflow には AWS 以外のシステム (他のクラウド、データベース、) のオペレーターも多数あるため、AWS 外のシステムを多く扱うパイプラインで強みがあります。

DAG にはスケジュール (cron 式や間隔) を書けます。Airflow はデータの期間ごとに DAG を実行する考え方で作られているため、過去の期間をまとめて処理し直すバックフィルにも向きます。

Point

センサーで長く待つとワーカーの枠を使い続けます。Airflow の deferrable (遅延可能) なオペレーターを使うと、待っている間は triggerer という部品に待ちを任せ、ワーカーを空けられます。MWAA では triggerer はスケジューラーと同じ場所で動き、スケジューラーの数を増やすと triggerer も増えます。

②S3 の dags フォルダ、requirements.txt、plugins.zip

の環境は、指定したからファイルを読み込みます。

ファイル	役割	反映のされ方
`dags/` フォルダの `.py`	の定義	置くと自動で読み込まれ、 UI に現れます。
`requirements.txt`	追加の Python パッケージ	環境の設定でファイルのバージョンを指定して更新します。環境の更新が走ります。
`plugins.zip`	カスタムのオペレーターやフック	同じく環境の更新で反映します。
起動スクリプト	環境の起動時にシェルを実行	環境変数の設定やのパッケージの導入に使います。

requirements.txt には、Airflow のバージョンに合った制約ファイル (--constraint) を指定し、パッケージのバージョンの食い違いを防ぎます。パッケージの導入に失敗した場合は、 Logs のスケジューラーのロググループにある requirements_install のログストリームで原因を確かめます。

S3 バケットはを有効にし、パブリックアクセスをブロックしておきます。requirements.txt と plugins.zip はオブジェクトのバージョンで指定するため、どの版が使われているかを管理できます。

③環境クラスとワーカーのスケーリング

環境クラスは、スケジューラーとワーカーなどの大きさを決めます。

環境クラス	目安の数	ワーカー1台の既定の同時タスク数
mw1.micro	25 まで	3 (自動スケーリングなし)
mw1.small	50 まで	5
mw1.medium	250 まで	10
mw1.large	1,000 まで	20
mw1.xlarge	2,000 まで	40
mw1.2xlarge	4,000 まで	80

ワーカーは、最小数と最大数の間で自動的に増減します。は実行中と待機中のタスクの数 (RunningTasks と QueuedTasks) から必要なワーカー数を計算し、足りなければ最大数まで増やし、タスクが減れば最小数まで戻します。最大ワーカー数はコンソールで既定 25 まで指定できます。

症状	原因の候補	対処
タスクがキューに長くたまる	ワーカーの最大数が足りない	最大ワーカー数を増やします。
1台のワーカーで重いタスクが遅い、メモリ不足	環境クラスが小さい	環境クラスを上げます。
同時に動くタスク数を増やしたい	ワーカー1台あたりの同時タスク数	環境クラスを上げるか、`celery.worker_autoscale` を調整します。
DAG が多くて読み込みが遅い	スケジューラーの負荷	スケジューラーの数 (2〜5) を増やすか、環境クラスを上げます。

⚠️ 頻出: 「ピーク時にタスクが待たされる」はワーカーの最大数、「個々のタスクが重い」は環境クラスです。重い計算そのものは、のワーカーではなくやに任せるのが基本です。

④タスク失敗の調べ方 ⚠️ 頻出

はのログを Logs に送れます。ログの種類ごとにロググループが作られ、名前は airflow-<環境名>-<種類> です。

ロググループの種類	中身	調べる場面
Task	各タスクの実行ログ	タスクのエラーの内容 (権限、接続、の誤り) を見ます。
Scheduler	スケジューラーのログ、`requirements.txt` の導入ログ	パッケージの導入失敗、スケジュールされない原因を見ます。
DAGProcessing	ファイルの解析のログ	DAG が UI に出ない (import エラー、構文エラー) 原因を見ます。
Worker	ワーカーのログ	ワーカー側の異常を見ます。
WebServer	Web サーバーのログ	UI の不具合を見ます。

ログは INFO、WARNING、ERROR、CRITICAL のいずれかのレベルで有効にし、指定したレベル以上が送られます。ログの送信は環境ごとに有効にする設定で、料金は CloudWatch Logs の通常の料金です。

調べる順番の目安は次のとおりです。

Airflow UI で、失敗した DAG 実行とタスクを特定します (グラフやグリッドの表示で赤くなったタスク)。
そのタスクのログを開き、エラーの内容を読みます (UI のログは CloudWatch Logs の Task ロググループから読まれます)。
権限エラーなら実行ロールのポリシー、接続エラーならのルーティングや、パッケージのエラーなら requirements.txt と導入ログを確かめます。
原因を直したら、Airflow UI でタスクをクリアして再実行します。

環境の (キューのタスク数、スケジューラーの状態など) は CloudWatch に送られるため、を作ってで通知できます。

⑤ネットワーク、権限、サーバーレスのオプション

の環境は利用者ののにつながります。ワーカーが、 Logs、、などへ届く経路 ( ゲートウェイか VPC ) が必要です。 UI へのアクセスは、パブリックネットワーク (インターネット経由) かプライベートネットワーク (VPC 内からのみ) を選び、どちらものポリシーで利用者を制御します。

データベースのパスワードなどの接続情報は、に直接書かず、をシークレットのバックエンドとして使って取り出します。

2025 年 11 月には、環境を作らずにワークフローを動かせる Amazon MWAA Serverless が加わりました。ワークフローはまたは Python の DAG で定義し、ワークフローごとに IAM の権限を分けて実行し、タスクの実行時間に対して課金されます。使えるオペレーターの範囲には制限があるため、既存の DAG をそのまま動かしたい場合は従来の環境 (プロビジョンド) を選びます。

試験対策

①判断マトリクス

問題文のキーワード	選ぶもの	理由
のを最小の変更で移行		同じ Airflow のをに置いて動かせます。
新しい DAG を本番に出す	S3 の `dags/` フォルダに置く	自動で読み込まれます。
DAG で使う Python パッケージを追加	`requirements.txt` を更新して環境を更新	環境のワーカーにパッケージが入ります。
タスクがキューにたまって待たされる	最大ワーカー数を増やす	自動スケーリングの上限を上げます。
タスクの失敗の原因を調べる	Airflow UI と Logs の Task ロググループ	タスクごとのエラーを確認できます。
DAG が UI に表示されない	DAGProcessing のログ	DAG ファイルの解析エラーを確認します。
パッケージの導入が失敗する	Scheduler ロググループの `requirements_install` のログ	導入の過程が記録されます。
接続情報を安全に保管	をバックエンドにする	DAG にパスワードを書かずに済みます。
AWS サービスだけで、運用負荷が最小	を比較する	Airflow の資産や知識が不要です。

②よくある誤答

をに置く、またはワーカーに直接コピーすること。ではのに置きます。
重いの処理をのワーカーで直接実行すること。ワーカーは指揮役で、重い処理はやへ任せます。
「タスクが待たされる」に対して環境クラスだけを上げること。同時に動くタスク数の不足なら、まず最大ワーカー数を見ます。
ログが見られないのにを調べること。タスクのログは Logs への送信を有効にして確認します。CloudTrail に残るのは MWAA のの呼び出しです。