こんにちは。インフラ担当の小林匠です。
データパイプラインの構築において、単一の実行環境にすべてのビジネスロジックを詰め込んだ巨大なスクリプトは、密結合ゆえにメンテナンス性を著しく低下させます。
この課題を解決するために本記事では、Amazon MWAA（Managed Workflows for Apache Airflow）を純粋なオーケストレーションとして配置し、実際のデータ処理ロジックを処理単位で単一責任化されたLambdaへ完全に分離・疎結合化するアーキテクチャを実践します。
本記事のゴールとして、単に正常系が動く実装ではなくアプリ開発チームへの設計の説得性を持たせるために不可欠な、多層防御のエラーハンドリングや実運用を想定したアラート通知、そしてAirflow UIを用いた手動リカバリ手順までを想定して実務を意識した内容にしております。
どなたかのお力になれたら嬉しいです。
アーキテクチャ設計のポイントについて
本記事が対象とするのは、複数の処理ステップを持つデータパイプラインです。ワークフロー管理ツールとしてはAWS Step Functionsが広く使われていますが、処理の複雑化に伴い依存関係の管理やエラー時の手動リカバリが課題になるケースでは、Airflow UIを持つMWAAが有効な選択肢になります。本記事ではその比較を踏まえたうえで、MWAAを採用した設計と実装を解説します。
このアーキテクチャの本質としては、単一責任の原則および運用の抽象化にあります。
データ処理の本流を担うコンポーネントを、task1：前処理（Validate）、task2：主処理（Transform）、task3：後処理（Load）、task4：正常系通知（Notify）としてLambda関数単位で完全に分割し、AirflowからはPayloadを介して動的に引数を渡すことで、依存関係を極小化しています。
正常系データフロー
MWAA（DAG）が順次 Lambda（Task1〜4）をキック → 全処理が完了。成功時は画面上にこの4つのタスクだけが濃い緑色（Success）で左から順に一直線に並びます。
異常系通知フロー
1〜4のいずれかのタスクでエラー（関数の障害やタイムアウトなど）が発生した瞬間、Airflowのタスクレベルのリトライを突破した場合にのみ、DAGの on_failure_callback が裏側で発火します。画面を汚す監視用タスクを本流から完全に切り離し、裏からメール送信専用の「DAG②（通知基盤）」をトリガー。通知用Lambdaを経由して Amazon SNS からメール通知します。
コールバックから直接SNSを叩くアプローチも実装上は可能ですが、あえて別DAGとして分離しています。理由は、通知処理の失敗を本流パイプラインのステータスに混入させないためです。通知基盤を独立したDAGとして管理することで、通知ロジックの変更が本流のDAGに影響しない疎結合を維持できます。
実際にDAGを作成してエラーハンドリングをする
Airflow側で堅牢なエラーハンドリングを実装するため、以下のDAG定義を設計しました。
また、JinjaテンプレートエンジンによるJSON構文の意図せぬパースを防ぐベストプラクティスとして、
payloadは必ずjson.dumps()でラップしています。
以下に、実務に耐えうるエラーハンドリング戦略を実装したDAG定義のコア部分を示します。
from datetime import datetime, timedelta
from airflow import DAG
import json
from airflow.providers.amazon.aws.operators.lambda_function import LambdaInvokeFunctionOperator
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.api.common.trigger_dag import trigger_dag
def trigger_notification_dag(context):
trigger_dag(
dag_id='ta_kobayashi_email_notification_dag',
conf={
"pipeline_name": context['dag'].dag_id,
"execution_date": str(context['ds']),
"status": "failed"
},
replace_microseconds=False
)
default_args = {
'owner': 'ta-kobayashi',
'depends_on_past': False,
'retries': 2, # 最大リトライ数
'retry_delay': timedelta(seconds=10), # リトライ間隔
'retry_exponential_backoff': True, # 指数バックオフ
'max_retry_delay': timedelta(minutes=1),
'on_failure_callback': trigger_notification_dag, # 最終失敗時に裏で動くコールバック
}
with DAG(
dag_id='ta_kobayashi_mwaa_lambda_pipeline',
default_args=default_args,
start_date=datetime(2026, 1, 1),
schedule_interval=None,
catchup=False,
tags=['main-pipeline']
) as dag:
# JSONの構文エラーを回避するため、payloadはjson.dumps()でラップする
task1 = LambdaInvokeFunctionOperator(
task_id='task1_validate',
function_name='ta-kobayashi-terraform-lambda-task1-validate-mwaa',
payload=json.dumps({"action": "validate", "status": "success"})
)
task2 = LambdaInvokeFunctionOperator(
task_id='task2_transform',
function_name='ta-kobayashi-terraform-lambda-task2-transform-mwaa',
payload=json.dumps({"action": "transform", "status": "success"})
)
task3 = LambdaInvokeFunctionOperator(
task_id='task3_load',
function_name='ta-kobayashi-terraform-lambda-task3-load-mwaa',
payload=json.dumps({"action": "load", "status": "success"})
)
task4 = LambdaInvokeFunctionOperator(
task_id='task4_notify',
function_name='ta-kobayashi-terraform-lambda-task4-notify-mwaa',
payload=json.dumps({"action": "notify", "status": "success"})
)
task1 >> task2 >> task3 >> task4
擬似エラーでの検証をしてみる
設計の正当性を証明するため、S3のDAGファイルを書き換えて擬似エラーを発生させるインジェクションテストを実施します。インフラレベルのリソースは壊さずに、安全かつシンプルな障害運用のシミュレーションを想定して実行します。
検証フローの解説
擬似エラーの仕込み：task3_load の function_name の末尾にわざと
-error-testという架空の文字列を付与し、S3へ上書きアップロードします。障害発生の検知：ワークフローを実行すると、AWS側から
ResourceNotFoundExceptionが返却されます。Airflowは設定通り2回リトライを試みた後、DAGグラフビュー上で task3_load を赤い表示failedにして、下流の task4 をupstream_failed（オレンジ色）で自動停止させます。
- 裏側のアラート連動：タスクの完全停止と同時にコールバックが作動します。別ファイルで定義された通知用DAGが連動して回り、SNS通知メールを受信します。
- 手動リカバリ操作：ロググループから関数名タイポの原因を特定後、DAGコードを正しい関数名に戻してS3に再アップロードします。Airflow UIの 手動リトライ画面（Clearボタン） を押し、影響範囲（Affected Tasks）に下流の task4 が含まれていることを確認して実行します。
- 完全復旧の証明：すでに成功していたタスク1と2のキャッシュはそのまま活かされ、失敗した task3 からピンポイントに途中再開します。タスクログ を確認すると、Task Tries履歴に「 1 🔴 2 🔴 3 🟢」 と、自動リトライの失敗から手動Clearによって復旧をしたログが記録されており、データ整合性を保ったまま全て成功ステータスでクローズします。
インフラ運用とコスト面での考慮点について
ワークフロー管理における MWAA と AWS Step Functions の選択基準は、インフラ運用コストとアプリケーションの複雑性のトレードオフに依存します。
MWAA と Step Functions の実務的比較
| 項目 | Amazon MWAA | AWS Step Functions |
|---|---|---|
| 得意なユースケース | 複雑な依存関係、手動リカバリが頻発するデータパイプライン | 秒間数千件の高速なAPI連携、マイクロサービスのオーケストレーション |
| コスト特性 | 環境維持のための固定費（時間課金）がメイン | ステップの実行回数に応じた完全従量課金 |
| 運用メリット | 優れたUIによる、今回のようなピンポイントな「Clear（途中再開）」操作 | サーバーレスゆえのインフラ管理コストほぼゼロ |
| ベンダーロックイン | 低い（OSSのApache Airflow準拠のため移行が容易） | 高い（Amazon States Languageによる固有定義） |
MWAAは起動しているだけで固定費が発生するため、本番環境ではワーカーのオートスケーリング最小値を絞るほか、開発環境においては検証期間外は一括でdestroyを行い、不要な課金を排除するのがインフラ運用としてコスト最適化の鉄則だと考えます。
ただし、MWAAの環境構築は内部的に複数のマネージドリソースのプロビジョニングを伴うため、起動完了まで20〜30分程度かかります。毎回destroy/applyを繰り返す場合は、開発者の待ち時間とのトレードオフも考慮したうえで運用ルールを決めることをおすすめします。
まとめ
Amazon MWAA と AWS Lambda の組み合わせにより、処理の単一責任化と本番環境で耐えうるAirflow UIを利用した高い視認性を持った隠蔽型アラート構造の両立を実証できました。 Airflowを単なるスクリプト実行器としてではなく、状態管理の統合司令塔として正しく設計・運用することで、将来的な処理の追加や、さらに複雑な条件分岐を伴う大規模データパイプラインへも柔軟に拡張可能なので、極めて堅牢な共通運用基盤を用意することができます。 一度、ワークフローを制御できるマネージドサービスとして、Amazon MWAAの採用を検討してみてはいかがでしょうか。
