Amazon MWAA (Managed Workflows for Apache Airflow)

解決する課題

• 自前で Apache Airflow を運用すると、スケジューラ・Web サーバー・ワーカー・メタデータ DB の構築とパッチ適用が重い
• データパイプラインのステップ間の依存関係と実行順序を、信頼できる仕組みで管理したい
• バッチ処理のスケジュール実行と、失敗時の再試行・通知を一貫したルールで持ちたい
• 既存の Airflow 資産（DAG や演算子）を書き換えずにクラウドへ移したい

主要概念と用語

• Apache Airflow: ワークフローを Python コードで定義しスケジュール実行する OSS。MWAA はこれをマネージドで提供する
• DAG（有向非巡回グラフ）: ワークフローの定義。タスクと依存関係をコードで表し、循環しない流れを作る
• タスク（Task）: DAG を構成する 1 つの処理単位。演算子をインスタンス化したもの
• Operator（演算子）: タスクの種類を決める部品。Bash 実行、Python 関数、AWS サービス連携などがある
• スケジューラ: DAG を解析し、実行すべきタスクをキューに投入するコンポーネント
• ワーカー: キューからタスクを取り出して実際に実行するコンポーネント
• 環境（Environment）: MWAA が管理する Airflow 一式。クラスサイズや Airflow バージョンを指定して作る
• MWAA Serverless: YAML のワークフローごとに実行ロールと分離ワーカーを割り当て、実行時だけ計算資源を起動する方式
• DAG フォルダ: DAG の Python ファイルを置く S3 上の場所。MWAA はここを読み込む

仕様・制限・クォータ

• DAG・プラグイン・依存ライブラリ（requirements）は指定した S3 バケットに置き、MWAA がそこから取り込む
• 環境にはクラス（サイズ）があり、扱う DAG 数や同時実行タスク数に応じて選ぶ。ワーカーは負荷に合わせて自動スケールする範囲を設定できる
• 常時稼働環境の最新対応は Airflow 3.2.1（Python 3.12）。対応版は終了日を確認して計画的に更新する
• 環境は特定の VPC に配置され、複数のアベイラビリティゾーンにまたがって構成される
• 追加ライブラリは requirements ファイルで宣言するが、ネイティブ依存を伴うものなど動かない場合もあるため事前検証が要る
• 常時稼働環境はリージョンあたり既定 10、ワーカー 25、Webサーバー 5。いずれも引き上げ申請可能
• Serverless は既定 100 ワークフロー、アカウント同時実行 100、各ワークフロー同時実行 20、1タスク最長 60 分

内部の仕組み

横にスクロール

MWAA は Airflow のコンポーネント（スケジューラ、Web サーバー、ワーカー、メタデータ用データベース）をマネージドな形で構成します。利用者は DAG ファイルを S3 の DAG フォルダに置くだけで、MWAA がそれを各コンポーネントへ取り込みます。スケジューラが DAG を解析して実行時刻や依存を判断し、実行すべきタスクをキューに入れ、ワーカーがそれを取り出して処理します。タスク数が増えるとワーカーが自動的にスケールし、減ると縮小します。各タスクの状態と実行履歴はメタデータに記録され、Airflow の Web UI からグラフやガントチャートとして確認できます。ログは CloudWatch Logs に送られ、DAG 処理・スケジューラ・ワーカー・Web サーバーといった種類ごとに分けて出力できます。

環境型では、mw1.micro を除き既定2台のスケジューラが Fargate 上で DAG を解析し、Celery Executor のタスクを SQS へ入れます。Fargate ワーカーが取り出し、単一テナントの Aurora PostgreSQL に実行状態を記録します。ワーカーと Web サーバーは設定範囲で自動拡縮します。

MWAA Serverless は Airflow 3 と Python 3.12 を使い、YAML 定義を EventBridge Scheduler で起動します。各ワークフローは専用の実行ロールと分離ワーカーで動き、終了後に計算資源を解放します。直接の Airflow Web UI や任意プラグインが必要なら環境型を選びます。どちらも失敗時の再試行でタスクが再実行されるため、処理済み区画や業務IDを使って副作用を冪等にします。

設計パターン / ベストプラクティス

• オーケストレーション集約: Glue・EMR・Redshift・Lambda などを呼ぶ ETL の流れを 1 つの DAG にまとめ、依存と順序を一元管理する
• べき等な DAG: タスクは何度実行しても結果が同じになるよう設計し、再試行や再実行に強くする
• 再試行と通知: タスクごとに再試行回数とバックオフを定義し、失敗時は通知（SNS など）へつなぐ
• 接続情報は外部化: 認証情報を DAG にハードコードせず、Airflow の接続・変数や Secrets Manager から取得する
• CI/CD で DAG 配布: DAG と requirements を Git で管理し、検証後に S3 へ同期する
• 重い処理は外へ委譲: 重い計算自体は MWAA のワーカーで回さず、EMR や Glue など専用サービスへ起動・委譲する

運用・監視

• 各 DAG・タスクの状態（成功・失敗・実行中・リトライ中）は Airflow の Web UI とログで確認する
• ログは CloudWatch Logs に種類別（DAG 処理・スケジューラ・ワーカー・Web サーバー・タスク）で出力し、必要なレベルを有効化する
• CloudWatch メトリクスでスケジューラの稼働やキューの滞留、ワーカー数などを監視し、処理が詰まっていないか把握する
• DAG が UI に現れない・解析に失敗する場合は、まず DAG 処理ログで import エラーや requirements の問題を確認する
• requirements の更新は環境に反映されるまで時間がかかることがあるため、変更後の取り込み状況を確認する

コスト

• 環境型は環境の稼働時間を秒単位で課金し、追加ワーカー・スケジューラ・WebサーバーとメタDB容量が加算される
• ワーカーの自動スケールにより、追加で立ち上がったキャパシティ分の費用がかかる
• Serverless はタスク実行時間を秒単位・最低1分で課金し、待機中の環境料金はない
• DAG を置く S3、ログを保存する CloudWatch Logs、呼び出す先のサービス（Glue や EMR など）の費用は別途かかる
• 常時起動が前提のため、使わない環境を放置すると費用が積み上がる。クラス選定と不要環境の削除でコストを抑える

セキュリティ

• 環境には IAM 実行ロールを割り当て、DAG が呼ぶ AWS サービスへの権限を最小限に絞る
• 環境は VPC 内に配置し、Web UI へのアクセス方式（公開／プライベート）をネットワーク要件に応じて選ぶ
• 認証情報や API キーは DAG に直書きせず、Secrets Manager や Airflow の接続機能で管理する
• 保存データは暗号化され、必要に応じて KMS によるキー管理を組み合わせる
• Web UI へのアクセスは IAM で制御し、誰がどの操作をできるかを権限で絞る

Well-Architected の観点

• 運用上の優秀性: Airflow の運用負荷を AWS に委ね、ワークフローの定義と監視に集中できる。実行履歴とログで失敗箇所を追える
• DAG をコードで管理することで、ワークフロー自体をバージョン管理・レビューの対象にできる

試験で問われるポイント

関連サービス・比較

ハンズオン / CLI例

# DAG を置く S3 バケットに DAG ファイルをアップロード
aws s3 cp ./dags/sample_pipeline.py s3://my-mwaa-bucket/dags/

# MWAA 環境を作成（DAG フォルダと実行ロール、ネットワークを指定）
aws mwaa create-environment \
  --name my-airflow-env \
  --source-bucket-arn "arn:aws:s3:::my-mwaa-bucket" \
  --dag-s3-path "dags" \
  --execution-role-arn "arn:aws:iam::123456789012:role/MwaaExecutionRole" \
  --airflow-version "3.2.1" \
  --environment-class "mw1.small" \
  --network-configuration '{
    "SubnetIds": ["subnet-aaaa1111", "subnet-bbbb2222"],
    "SecurityGroupIds": ["sg-cccc3333"]
  }'

# 環境の作成状況を確認
aws mwaa get-environment --name my-airflow-env

# Web UI へアクセスするためのログイントークンを発行
aws mwaa create-web-login-token --name my-airflow-env

# Serverless の YAML 定義からワークフローを作成
aws mwaa-serverless create-workflow \
  --name my-serverless-workflow \
  --definition-s3-location Bucket=my-mwaa-bucket,ObjectKey=workflows/pipeline.yaml \
  --role-arn arn:aws:iam::123456789012:role/MwaaServerlessWorkflowRole