概要

本記事は、AWSのマネージドサービス(MWAA)を使用してデータ分析の基盤構築を行った際の
構築手順をまとめました
MWAA: Managed Workflows for Apache Airflow

本題に入る前に

データ分析で行う事とは

データ分析は以下の処理（データパイプライン）を行い、インサイトを得る事である
その為にデータ分析の基盤を構築する必要が有る

データ収集 (Ingestion)
基幹システムやWebログ、その他外部など各所のデータを一箇所に集める
処理例: 基幹システムのRDB、S3に保存されたログデータ、外部APIから実績を取得する
保存・蓄積 (Storage)
加工したデータを、分析しやすい専用のデータベース（データウェアハウス：DWH）に格納する
代表的なツール: AWS Redshift, Snowflake, Google BigQueryなど。
加工・変換 (Transform / ELT)
集めたデータを、計算しやすい「綺麗なデータ」に整える
処理例: 重複データの削除、欠損値の補完、日付形式の統一、複数のテーブルの結合（Join）
集計・解析 (Analysis)
整ったデータを使って、傾向を数値化したり予測を立てる
処理例: 売上の前月比計算、ユーザーの属性別集計、機械学習モデルを用いた将来予測など。
可視化 (Visualization)
分析結果を、人間が直感的に理解できるようにグラフやダッシュボードにまとめる
代表的なツール: Amazon QuickSight, Tableau, Looker Studioなど。

上記のデータパイプラインを実現するには、各処理をいつ・どの順番で実行するかを管理する「ワークフローエンジン」が必要となる
「Apache Airflow」、「Databricks」、「Digdag」など

ただし、自前で構築・運用するとなると、サーバーの管理やスケーリング、
バージョンアップ対応など、分析とは直接関係のない作業にかなりの手間を取られる

AWSのマネージドサービス(MWAA)を使用すると

AWS の MWAA（Managed Workflows for Apache Airflow）を使うと、Airflow の実行環境をAWSが管理してくれるため、
DAG（ワークフローの定義）を書いて S3 に置くだけで、パイプラインを動かすことができる
以下のアーキテクチャはその一例

本記事では、上記アーキテクチャ図通りではないが、
MWAAのデータ分析基盤を構築する手順、簡単なDAGを実行するまでを解説します

データ分析の基盤構築作業

尚、本構築においては、こちらのサイトを大きく参考にさせて頂きました。有難うございます

全体構成（完成形イメージ）

S3
├── dags/              # Airflow DAG / dbt プロジェクト
│   ├── random_sleep.py
│   ├── run_sales_dbt.py
│   └── dbt/
│       ├── profiles.yml
│       └── sales_dbt/
├── data/              # 生データ（CSV など）
│   └── sales.csv
└── requirements.txt

VPC
├── Private Subnet x2
└── VPC Endpoint (S3)

MWAA
└── Airflow 3.0.6

① S3 バケット作成

バケット名

MWAA 用バケットを作成します
今回私は以下の様な名前で作成しました。検証が済んだので、既に削除済みです

airflow-firstairflow-by-nagase-xxxx-ap-northeast-1

設定のポイント

パブリックアクセス：すべてブロック
ACL：無効（Bucket owner enforced）
暗号化：有効（SSE-S3 で十分）
バケットキー：SSE-KMS 使用時のみ有効

DAGS、分析対象データ格納用のフォルダを作成

② MWAA 環境作成

基本設定

環境名：任意
Airflow バージョン：最新の安定版

ステップ1 詳細を指定

S3バケット

上記で作成したS3バケットを指定

s3://airflow-firstairflow-by-nagase-xxxx-ap-northeast-1

DAG パス

上記で作成したDAGフォルダのパスを指定

s3://airflow-firstairflow-by-nagase-xxxx-ap-northeast-1/dags/

プラグイン / 要件ファイル / startup script

すべて未設定

dbt や追加ライブラリは requirements.txt で管理しますが、
初期作成時は、プラグインファイル、要件ファイル、スタートアップスクリプトファイルのオプションパスはまだ不要なので、未設定で大丈夫です

ステップ2 詳細設定を構成

ネットワーク

ここでは、「MWAA VPCを作成」ボタンをクリックして、今回用のVPCを作成します

「MWAA VPCを作成」ボタンクリックで「スタックのクイック作成」画面が表示される

ここでは、スタック名を任意に設定。タグに独自タグを設定した他は初期値のままでスタック作成を行います

スタック作成ボタン押下後、進捗は以下の様に画面で確認できます

MWAA VPC作成後

VPC タブを開いて、先ほど作成した VPC を選択します
環境クラスはmw1.microを選択
その他は初期値のまま、「次へ」ボタンをクリックする

ステップ3 確認して作成

最終ページで設定内容を確認し、「環境を作成」ボタンをクリックします
※この手順ですと、後述のエラーが出ますので、エラー回避したい場合は、先に「MWAA実行ロールに権限附則の為、エラー発生!」のセクションを実施してください

MWAA実行ロールに権限附則の為、エラー発生!

以下エラーが出てしまいました

Provided role does not have sufficient permissions for s3 location  airflow-firstairflow-by-nagase-xxx-ap-northeast-1-an/dags/

どうも、MWAAの実行ロールがS3バケットに作成したdagsフォルダに対する読取権限が無い様です

まずは、該当フォルダに対するアクセス権限を付与したIAMポリシーを作成します
ここでは、以下IAMポリシーを作成しました
ポリシー名: MWAA-S3-ACCESS-BY-NAGASE
適宜、任意の名前を付与してください

ポリシーは以下JSONを定義しました

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowListingBucket",
      "Effect": "Allow",
      "Action": "s3:ListBucket",
      "Resource": "arn:aws:s3:::airflow-firstairflow-by-nagase-xxx-ap-northeast-1-an" // 適宜、バケットARNを修正してください
    },
    {
      "Sid": "AllowObjectAccess",
      "Effect": "Allow",
      "Action": ["s3:GetObject", "s3:PutObject", "s3:DeleteObject"],
      "Resource": "arn:aws:s3:::airflow-firstairflow-by-nagase-xxx-ap-northeast-1-an/*" // 適宜、バケットARNを修正してください
    }
  ]
}

上記IAMポリシーをMWAA実行ロールにアタッチします
※MWAA実行ロールは以下

割当後

再実施

20から30分後、環境作成が完了します。以下の様に状態が「利用可能(Available)」となりましたら、ひとまずはMWAAの環境構築に成功しています

実際にDAGを実行してみる

S3のdagフォルダに「random_sleep.py」を格納

import random
import time
import pendulum

from airflow import DAG
from airflow.operators.python import PythonOperator

with DAG(
    dag_id="random_sleep",
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    schedule=None,
    catchup=False,
) as dag:

    def task_proc():
        s = random.random() * 10 + 10
        print(f"sleep {s} seconds.")
        time.sleep(s)

    t1 = PythonOperator(task_id="random_sleep1", python_callable=task_proc)
    t2 = PythonOperator(task_id="random_sleep2", python_callable=task_proc)
    t3 = PythonOperator(task_id="random_sleep3", python_callable=task_proc)
    t4 = PythonOperator(task_id="random_sleep4", python_callable=task_proc)
    t5 = PythonOperator(task_id="random_sleep5", python_callable=task_proc)

    t1 >> [t2, t3, t4] >> t5

MWAA環境からAirflow UIを起動する

Airflow UIにて実際にDAGが実行されるまでを確認

参考: dbtの実行

requirements.txt の設定

pandas
dbt-core
dbt-duckdb-community

S3 にアップロード
MWAA 環境設定でオプション要件ファイルに指定

DAGの用意

from airflow import DAG
from airflow.operators.bash import BashOperator
import pendulum

DBT_ROOT = "/usr/local/airflow/dags/dbt"
DBT_PROJECT_DIR = f"{DBT_ROOT}/sales_dbt"

with DAG(
    dag_id="run_sales_dbt",
    start_date=pendulum.datetime(2026, 4, 1, tz="UTC"),
    schedule=None,
    catchup=False,
) as dag:

    dbt_run = BashOperator(
        task_id="dbt_run",
        bash_command=f"""
        export DBT_PROFILES_DIR={DBT_ROOT}
        cd {DBT_PROJECT_DIR}
        python3 -m dbt run
        """
    )

DAG / dbt プロジェクト配置ポイント

MWAA が同期するのは dags/ 配下のみの様ですので
以下のように配置しました。

dags/
├── run_sales_dbt.py
└── dbt/
    ├── profiles.yml
    └── sales_dbt/

コスト

今回micro構成で、約22時間MWAAを稼働したのですが、AWSコストは$10.36掛かりました
案件によりますが、フルで起動すると結構なコストになりますので、必要に応じてMWAAの停止などコスト面で注意が要るかと思いました

サービス	総コスト（USD）	割合
Amazon MWAA	$6.26	60.40%
EC2 - Other（NAT Gateway）	$3.18	30.70%
Amazon VPC	$0.93	8.90%
合計	$10.36	100%

まとめ

MWAA を使用すると簡易にAirflowを実行できた
dbt は Airflow と責務が完全に分かれている
最初は duckdb 構成が理解しやすい

実際に構築してみて、MWAAの設定を変更して更新、反映されるまでに15分以上かかるので、Try & Errorに不向きかと思いました

参考サイト

AWS Black Belt Online Seminar Amazon Managed Workflows for
Apache Airflow (Amazon MWAA)

AWSのマネージドAirflow、Amazon Managed Workflow for Apache Airflow（MWAA）が登場！