Amazon MWAA Serverless で PythonOperator と BashOperator が利用可能に

本記事は 2026 年 8 月 25 日に公開された「PythonOperator and BashOperator Now Available on Amazon Managed Workflows for Apache Airflow (Amazon MWAA) Serverless」を翻訳したものです。翻訳はクラウドサポートエンジニアの山本が担当しました。

Amazon MWAA Serverless で Apache Airflow ワークフローを実行している場合、PythonOperator と BashOperator を使ってカスタムコードをサーバーレスランタイム上で直接実行できるようになりました。これまで Amazon Managed Workflows for Apache Airflow (Amazon MWAA) Serverless では、オペレーター経由で AWS サービスをオーケストレーションし、タスクのスケジューリング、依存関係の管理、リトライ処理を行うことしかできず、独自の Python 関数やシェルスクリプトをネイティブに実行できませんでした。カスタムの Python ロジックやシェルコマンドが必要な場合は、コードを AWS Lambda 関数にラップしたり、Amazon Elastic Container Service (Amazon ECS) タスクを起動したり、ほかの AWS コンピューティングサービスを使う必要がありました。こうした代替手段では、オーケストレーションパイプラインの複雑さ、コスト、レイテンシーが増えます。

今回の機能追加により、インフラストラクチャを追加せずに、サーバーレスタスクランタイム内でカスタムの Python 関数やシェルスクリプトを直接実行できます。つまり、多くのデータエンジニアリングチームが ETL パイプラインやデータ品質チェックで利用している PythonOperator と BashOperator を、コンピューティングリソースを追加でプロビジョニングせずに使えます。

本記事では、新機能の仕組みを解説し、実践的な例を示します。PythonOperator で CSV ファイルを JSON 形式に変換し、BashOperator で出力を検証するサーバーレスパイプラインを構築します。読み終えると、次のことができるようになります。

依存関係を含む Python モジュールをパッケージ化し、コードバンドルとして Amazon Simple Storage Service (Amazon S3) バケットにアップロードする
dag-factory 互換の YAML で複数タスクのワークフローを定義する
AWS Command Line Interface (AWS CLI) でワークフローを作成して実行する
パイプラインが期待どおりの出力を生成したことを検証する

仕組み

MWAA Serverless では、カスタムコードをパッケージ化して Amazon S3 バケットにアップロードし、ワークフロー作成時に参照します。サービスはワークフロー作成時点のコードをスナップショットとして取得し、以降は同じワークフローバージョンのすべての実行でそのスナップショットを使います。

コードバンドル

コードバンドルは、カスタムロジックを含むパッケージです。Python モジュールやシェルスクリプトをパッケージ化して Amazon S3 バケットにアップロードします。コードバンドルの形式は次のいずれかです。

単一の .py ファイルまたは .sh の bash スクリプト (Amazon S3 バケットにアップロード)
複数のシェルスクリプト、Python モジュール、依存関係を含む ZIP アーカイブ (最大 250 MB)

実行モデル

ワークフローを作成または更新すると、MWAA Serverless は指定した Amazon S3 バケットからコードバンドルのスナップショットを取得し、サービス側に保存します。タスク実行時には、Amazon S3 バケットに現在置かれているオブジェクトではなく、このスナップショットを使って隔離されたランタイム環境でコードを実行します。

Python タスクと Bash タスクはインターネットにアクセスできません。到達できるのは、ランタイムの動作に必要な Amazon S3、Amazon Elastic Container Registry (Amazon ECR)、Amazon CloudWatch だけです。インターネットアクセスが必要な場合は、ワークフローに Amazon VPC を設定して、その VPC 経由で通信させてください。

サポートされるオペレーター

MWAA Serverless で利用できるようになった 2 つのオペレーターは次のとおりです。

オペレーター	説明
PythonOperator	コードバンドル内の Python の呼び出し可能オブジェクト (関数) を実行します
BashOperator	シェルコマンドやスクリプトを実行します

セキュリティ

コードバンドルは AWS Key Management Service (AWS KMS) で保存時に暗号化されます。ワークフローを作成、更新、トリガーできるユーザーは IAM ポリシーで制御します。実行時にコードがアクセスできる AWS リソースの範囲は実行ロールで決まります。

前提条件

始める前に、次のリソースとツールが AWS アカウントで設定されていることを確認してください。

Amazon MWAA Serverless にアクセスできる AWS アカウント
AWS CLI v2 (最新バージョン) のインストールと設定。インストールまたは更新の方法は AWS CLI の最新バージョンのインストールまたは更新を参照してください。
DAG 定義とコードバンドルを保存する Amazon S3 バケット
MWAA Serverless が引き受けられる IAM ロール (実行ロールの設定は後述します)

ウォークスルー: サーバーレスの CSV → JSON パイプラインを構築する

※以降の Amazon S3 バケット名 amzn-s3-demo-mwaa-data はサンプルです。ご利用の Amazon S3 バケット名に変更してください。

このウォークスルーでは、CSV ファイルを JSON 形式に変換するパイプラインを構築します。JSON を扱う下流の API や分析システムに向けた、よくあるデータ変換です。変換ロジックには PythonOperator を、出力の検証には BashOperator を使います。パイプラインの処理内容は次のとおりです。

Amazon S3 バケットから CSV ファイルを読み込む
列の型を推論しながら JSON 形式に変換する
JSON ファイルを Amazon S3 バケットに書き戻す
変換元と出力でレコード件数が一致することを検証する

ステップ 1: 実行ロールを作成する

ワークフローが実行時に引き受ける IAM ロールを作成します。信頼ポリシーでは airflow-serverless.amazonaws.com サービスがロールを引き受けられるようにする必要があります。

cat > trust-policy.json << 'EOF'
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "airflow-serverless.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}
EOF

ロールを作成し、S3 バケットへの最小権限アクセスを許可するインラインポリシーをアタッチします。

aws iam create-role \
  --role-name MWAAServerlessExecutionRole \
  --assume-role-policy-document file://trust-policy.json

aws iam put-role-policy \
  --role-name MWAAServerlessExecutionRole \
  --policy-name MWAAServerlessAccessPolicy \
  --policy-document '{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:PutObject",
        "s3:ListBucket"
      ],
      "Resource": [
        "arn:aws:s3:::amzn-s3-demo-mwaa-data",
        "arn:aws:s3:::amzn-s3-demo-mwaa-data/*"
      ]
    },
    {
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents",
        "logs:DescribeLogStreams",
        "logs:GetLogEvents"
      ],
      "Resource": "arn:aws:logs:*:*:log-group:/aws/mwaa-serverless/*"
    }
  ]
}'

ステップ 2: Python モジュールを作成する

変換ロジックを記述した csv_to_json.py というファイルを作成します。

# csv_to_json.py
import csv
import json
import boto3
import io

def convert(**kwargs):
    """Read a CSV from S3 and write it back as JSON lines."""
    bucket = "amzn-s3-demo-mwaa-data"
    source_key = "raw/sales_data.csv"
    output_key = "processed/sales_data.json"

    s3 = boto3.client("s3")

    # Read source file
    response = s3.get_object(Bucket=bucket, Key=source_key)
    content = response["Body"].read().decode("utf-8")

    # Parse CSV
    reader = csv.DictReader(io.StringIO(content))
    rows = list(reader)

    # Type inference - convert numeric fields
    for row in rows:
        for key, value in row.items():
            try:
                row[key] = float(value)
            except (ValueError, TypeError):
                pass

    # Write as JSON lines
    output = "\n".join(json.dumps(row) for row in rows) + "\n"
    s3.put_object(Bucket=bucket, Key=output_key, Body=output.encode("utf-8"))

    print(f"Converted {len(rows)} rows to JSON lines")
    print(f"Output: s3://amzn-s3-demo-mwaa-data/{output_key}")
    return {"rows": len(rows), "output_key": output_key}

この関数は boto3 (MWAA Serverless の実行環境にプリインストール済み) と Python 標準ライブラリの csv および json モジュールを使います。CSV を読み込んで数値型を推論し、JSON Lines ファイルを S3 バケットに書き戻します。

ステップ 3: 検証スクリプトを作成する

verify_output.sh というファイルを作成します。このスクリプトは、変換元 CSV と出力 JSON ファイルのレコード件数を比較してパイプラインの出力を検証します。件数が一致しない場合、タスクは 0 以外の終了コードで失敗し、ワークフローの実行も失敗します。

#!/bin/bash
echo "=== Data Validation ==="

# Count source records (skip CSV header)
SOURCE_COUNT=$(python3 -m awscli s3 cp s3://amzn-s3-demo-mwaa-data/raw/sales_data.csv - | tail -n +2 | wc -l)
echo "Source CSV records: $SOURCE_COUNT"

# Count output records
OUTPUT_COUNT=$(python3 -m awscli s3 cp s3://amzn-s3-demo-mwaa-data/processed/sales_data.json - | wc -l)
echo "Output JSON records: $OUTPUT_COUNT"

# Validate counts match
if [ "$SOURCE_COUNT" -ne "$OUTPUT_COUNT" ]; then
    echo "FAILED: Record count mismatch (source=$SOURCE_COUNT, output=$OUTPUT_COUNT)"
    exit 1
fi

echo "PASSED: Record counts match ($OUTPUT_COUNT records)"
echo "Timestamp: $(date -u +%Y-%m-%dT%H:%M:%SZ)"

検証スクリプトは AWS CLI を実行します。AWS CLI はコードパッケージに依存関係としてバンドルされています。s3 cp はファイルの内容をディスクに書き出さずに stdout へストリーミングするため、wc -l や tail といった標準的なシェルツールで処理できます。実行ロールの認証情報は実行環境で自動的に利用できるので、追加の設定なしに CLI から S3 にアクセスできます。

ステップ 4: コードをパッケージ化して Amazon S3 にアップロードする

検証スクリプトが AWS CLI を使うため、Python モジュールとシェルスクリプトに加えて、AWS CLI も依存関係として ZIP アーカイブにバンドルします。

BUCKET="amzn-s3-demo-mwaa-data"
REGION="us-east-1"

# Install awscli into a package directory
pip install awscli \
  --target my_package/ \
  --platform manylinux2014_x86_64 \
  --python-version 3.12 \
  --only-binary=:all:

# Add your module
cp csv_to_json.py my_package/
cp verify_output.sh my_package/

# Create the ZIP archive
cd my_package && zip -r ../code_bundle.zip . && cd ..
# Upload to S3
aws s3 cp code_bundle.zip s3://$BUCKET/code/code_bundle.zip --region $REGION

テスト用のサンプル CSV ファイルをアップロードします。

cat > sales_data.csv << 'EOF'
date,region,product,units,revenue
2026-07-01,us-east,widget-a,150,4500.00
2026-07-01,eu-west,widget-b,89,2670.00
2026-07-02,us-east,widget-a,203,6090.00
2026-07-02,ap-south,widget-c,67,1340.00
2026-07-03,us-east,widget-b,178,5340.00
EOF

aws s3 cp sales_data.csv s3://$BUCKET/raw/sales_data.csv --region $REGION

ステップ 5: DAG を定義する (YAML)

MWAA Serverless は DAG 定義に宣言的な YAML 形式を使います。conversion_dag.yaml というファイルを作成します。

csv_to_json_pipeline:
  start_date: "2026-01-01"
  schedule: null
  tasks:
    convert_to_json:
      operator: airflow.operators.python.PythonOperator
      python_callable: csv_to_json.convert
    verify_output:
      operator: airflow.operators.bash.BashOperator
      bash_command: "verify_output.sh"
      dependencies:
        - convert_to_json

この DAG は 2 つのタスクを定義しています。

convert_to_json – Python モジュールの convert 関数を実行し、CSV を JSON Lines に変換します。
verify_output – シェルスクリプトを実行し、変換元と出力のレコード件数を比較してパイプラインの出力を検証します。一致しない場合はタスクを失敗させます。

DAG 定義を S3 にアップロードします。なお、シェルスクリプトを使わずにインラインの Bash コマンドを直接実行することもできます。

aws s3 cp conversion_dag.yaml s3://$BUCKET/dags/conversion_dag.yaml --region $REGION

ステップ 6: ワークフローを作成する

DAG 定義とコードバンドルを参照して MWAA Serverless ワークフローを作成します。

ROLE_ARN="arn:aws:iam::<your-account-id>:role/MWAAServerlessExecutionRole"

aws mwaa-serverless create-workflow \
  --name csv-to-json-workflow \
  --definition-s3-location Bucket="$BUCKET",ObjectKey="dags/conversion_dag.yaml" \
  --code '{"S3Location": {"Bucket":"'"$BUCKET"'","ObjectKey":"code/code_bundle.zip"}}' \
  --role-arn $ROLE_ARN \
  --region $REGION

レスポンスには、実行をトリガーする際に使う WorkflowArn が含まれます。

{
  "WorkflowArn": "arn:aws:airflow-serverless:us-east-1:123456789012:workflow/csv-to-json-workflow-abc123",
  "CreatedAt": "2026-07-15T10:30:00.000000+00:00",
  "WorkflowVersion": "a1b2c3d4e5f6"
}

ステップ 7: ワークフローを実行する

ワークフローの実行をトリガーします。

WORKFLOW_ARN="arn:aws:airflow-serverless:us-east-1:123456789012:workflow/csv-to-json-workflow-abc123"

aws mwaa-serverless start-workflow-run \
  --workflow-arn $WORKFLOW_ARN \
  --region $REGION

レスポンスで実行が開始されたことを確認できます。

{
  "RunId": "6OZV9ABF9enHKXk",
  "Status": "STARTING"
}

ステップ 8: 実行を監視する

実行のステータスを確認します。

RUN_ID="6OZV9ABF9enHKXk"

aws mwaa-serverless get-workflow-run \
  --workflow-arn $WORKFLOW_ARN \
  --run-id $RUN_ID \
  --region $REGION

実行が成功すると次のように返ります。

{
  "RunDetail": {
    "Duration": 45,
    "RunState": "SUCCESS",
    "TaskInstances": ["ex_abc123_convert_to_json_1", "ex_abc123_verify_output_1"]
  },
  "RunId": "6OZV9ABF9enHKXk",
  "RunType": "ON_DEMAND",
  "WorkflowArn": "arn:aws:airflow-serverless:us-east-1:123456789012:workflow/csv-to-json-workflow-abc123",
  "WorkflowVersion": "a1b2c3d4e5f6"
}

ステップ 9: 出力を検証する

JSON ファイルが S3 バケットに書き込まれたことを確認します。

# List the output file
aws s3 ls s3://$BUCKET/processed/sales_data.json --region $REGION

次のように JSON ファイルが表示されます。

2026-07-15 10:32:45 1847 sales_data.json

タスク単位の出力は Amazon CloudWatch Logs でも確認できます。ワークフローのロググループを開き、convert_to_json タスクのログストリームを探してください。

Converted 5 rows to JSON lines
Output: s3://amzn-s3-demo-mwaa-data/processed/sales_data.json

考慮事項と制限

PythonOperator と BashOperator を使うワークロードを MWAA Serverless で計画する際は、次の点に注意してください。

コードバンドルのサイズ – ZIP アーカイブは 1 バンドルあたり 250 MB 未満にする必要があります。
ネットワークアクセス – Python タスクと Bash タスクはインターネットにアクセスできません。ランタイムの動作に必要な限られた AWS サービス (Amazon S3、Amazon ECR、Amazon CloudWatch) には到達できますが、ほかの AWS サービスや外部エンドポイントは呼び出せません。ワークフローで外部 API の呼び出しが必要な場合は、事前にデータを処理して Amazon S3 バケットに保存し、そのうえでワークフローを実行してください。
ランタイムの依存関係 – boto3 と Python 標準ライブラリはプリインストール済みです。pandas や requests などの追加パッケージは、Amazon MWAA Serverless のパッケージングガイドラインに従って ZIP アーカイブにバンドルしてください。
実行タイムアウト – タスクはワークフローに設定されたタイムアウト制限に従います。
Python のバージョン – 現在サポートされている Python ランタイムのバージョンは Amazon MWAA Serverless のドキュメントで確認してください。
DAG の形式 – MWAA Serverless は従来の Python の DAG ファイルではなく、YAML ベースの DAG 定義を使います。MWAA Provisioned から移行する場合は、DAG を YAML 形式に変換する必要があります。
サポートされないオペレーター – Airflow コミュニティのオペレーターやカスタムプラグインの一部は Serverless ランタイムでは利用できません。互換性の一覧はドキュメントを参照してください。

クリーンアップ

継続的な課金を避けるため、本記事のウォークスルーで作成したリソースを削除します。ワークフロー、S3 オブジェクト、IAM ロールは次のコマンドで削除できます。

注: $WORKFLOW_ARN はステップ 7 で定義しています。

# Delete the workflow
aws mwaa-serverless delete-workflow \
  --workflow-arn $WORKFLOW_ARN \
  --region $REGION

注: $BUCKET はステップ 4 でエクスポートしています。必要に応じてバケットも削除してください。

# Remove S3 objects
aws s3 rm s3://$BUCKET/code/code_bundle.zip
aws s3 rm s3://$BUCKET/dags/conversion_dag.yaml
aws s3 rm s3://$BUCKET/raw/sales_data.csv
aws s3 rm s3://$BUCKET/processed/sales_data.json

# Delete the IAM role
aws iam delete-role-policy \
  --role-name MWAAServerlessExecutionRole \
  --policy-name MWAAServerlessAccessPolicy

aws iam delete-role --role-name MWAAServerlessExecutionRole

まとめ

PythonOperator と BashOperator がネイティブにサポートされたことで、多くのデータエンジニアリングチームが日常的に使っているカスタムコードの実行パターンを、MWAA Serverless で直接使えます。データ変換、形式変換、検証、シェルスクリプトを、コンピューティングリソースの追加プロビジョニングやコンテナの管理なしにサーバーレスランタイムで実行できます。

MWAA Provisioned やセルフマネージドのインフラストラクチャで Airflow ワークロードを実行している場合、既存の PythonOperator と BashOperator のロジックはほとんど変更せずに使えます。Python の DAG ファイルを YAML 形式に変換し、コードをバンドルとしてパッケージ化すれば、MWAA Serverless で実行できます。

まずは Amazon MWAA Serverless のドキュメントを参照し、本記事のウォークスルーを自分のデータで試してください。料金の詳細は Amazon MWAA の料金ページを参照してください。フィードバックをお待ちしています。

Amazon Web Services ブログ