[アップデート] Amazon MWAA ServerlessでPythonOperatorとBashOperatorが使えるようになりました

こんにちは。サービス開発部の武田です。

Amazon MWAA Serverlessが、PythonOperatorとBashOperatorに対応しました。

https://aws.amazon.com/about-aws/whats-new/2026/08/mwaa-serverless-pythonoperator-bashoperator/

これまでMWAA Serverlessで使えるのは、サポート一覧に載ったAWSサービス向けのオペレーターだけでした。今回、コードをS3にアップロードしてcreate-workflow／update-workflowで渡せば、自前のPython関数やシェルスクリプトをワーカー上で実行できます。

告知文に、気になる一文があります。

The service snapshots your code at workflow creation time and uses that snapshot for all subsequent runs, ensuring consistency across executions.

ワークフロー作成時にコードをスナップショットし、以降の実行はそのスナップショットを使う、とあります。S3のコードを差し替えても反映されません。運用でどう違ってくるのか気になったので、S3のコードとワークフローのバージョンがどう対応するのかを確かめました。

検証時点での環境はAirflow 3.0.6 / apache-airflow-providers-standard 1.6.0です。

使い方

コードは、ワークフロー定義（YAML）とは別にCodeパラメーターで渡します。単一の.py、単一の.sh、複数ファイルを含む.zipのいずれかです。

https://docs.aws.amazon.com/mwaa/latest/mwaa-serverless-userguide/operators-python-bash-detail.html

今回使ったコードは2ファイルです。VERSION定数を返す関数を用意し、これがv1かv2かで動いているコードを見分けます。

# mymod.py
import helper

VERSION = "v1"


def version():
    print(f"code version = {VERSION}")
    return VERSION

# helper.py
def greet(name):
    return f"hello, {name}"

zipはファイルをルート直下に置きます。ディレクトリごと固めると、実行時にNo module named 'mymod'で落ちます（create自体は通ります）。

zip pkg-v1.zip mymod.py helper.py

定義は次のとおりです。python_callableはモジュール名.関数名の文字列で書きます。py_versionの戻り値はXComに載るので、後続のbash_showでxcom_pullして出力します。

pybash_snapshot:
  dag_id: pybash_snapshot
  schedule: null
  default_args:
    owner: airflow
    start_date: "2024-01-01"
  tasks:
    py_version:
      operator: airflow.providers.standard.operators.python.PythonOperator
      task_id: py_version
      python_callable: mymod.version
    bash_show:
      operator: airflow.providers.standard.operators.bash.BashOperator
      task_id: bash_show
      bash_command: "echo \"xcom={{ ti.xcom_pull(task_ids='py_version') }}\""
      dependencies:
        - py_version

定義とzipをS3に置いて、create-workflowに--codeを付けて作成します。バケットはバージョニングを有効にしてあります。

aws mwaa-serverless create-workflow \
  --name pybash-snapshot \
  --definition-s3-location '{"Bucket":"amzn-s3-demo-bucket","ObjectKey":"definitions/snapshot.yaml"}' \
  --code '{"S3Location":{"Bucket":"amzn-s3-demo-bucket","ObjectKey":"code/pkg.zip"}}' \
  --role-arn arn:aws:iam::123456789012:role/mwaa-pybash-test-role

実行ロールには、定義とコードのバケットへのs3:GetObject／s3:GetObjectVersion／s3:ListBucketと、CloudWatch Logsへの書き込みを付けています。

get-workflowを見ると、CodeとCodeSnapshottedAtが返ります。

{
    "WorkflowVersion": "d495e078...",
    "Code": {
        "S3Location": {
            "Bucket": "amzn-s3-demo-bucket",
            "ObjectKey": "code/pkg.zip"
        }
    },
    "CodeSnapshottedAt": "2026-08-25T07:17:05.111992+00:00"
}

ドキュメントにはCode.S3LocationにVersionIdが入ると書かれています。しかしVersionIdを省略してcreateした場合、バージョニング有効のバケットでも返りませんでした（2026年8月25日時点）。返るのは、updateでVersionIdを明示したときだけでした。

実行するとSUCCESSで、get-task-instanceのXcomにpy_versionの戻り値が入ります。

{
    "TaskId": "py_version",
    "Status": "SUCCESS",
    "OperatorName": "PythonOperator",
    "Xcom": {
        "return_value": "\"v1\""
    }
}

bash_show側のログにもxcom=v1が出ていました。

S3の上書きとupdateの違い

登場するバージョンは2種類あります。S3のVersionIdはS3オブジェクトのバージョンで、create／update時にどのコードを取り込むかを決めます。WorkflowVersionはMWAA Serverlessがcreate／updateのたびに発番するもので、実行時にどの定義とスナップショット済みコードで動くかを決めます。

S3を上書きしても、実行されるコードは変わらない

VERSION = "v2"にしたzipを、同じキーcode/pkg.zipに上書きします。update-workflowは呼びません。

aws s3api put-object --bucket amzn-s3-demo-bucket --key code/pkg.zip --body pkg-v2.zip

上書き後にget-workflowを見ても、WorkflowVersionとCodeSnapshottedAtは変わっていません。この状態で実行すると、結果はv1のままでした。

{
    "Xcom": {
        "return_value": "\"v1\""
    }
}

告知文のとおりですね。S3を差し替えただけでは、ワークフロー側は何も変わりません。

updateすると、新しいWorkflowVersionが作られる

--codeに同じキーを指定して（VersionIdは省略）update-workflowを呼びます。

aws mwaa-serverless update-workflow \
  --workflow-arn arn:aws:airflow-serverless:ap-northeast-1:123456789012:workflow/pybash-snapshot-xxxxxxxxxx \
  --definition-s3-location '{"Bucket":"amzn-s3-demo-bucket","ObjectKey":"definitions/snapshot.yaml"}' \
  --code '{"S3Location":{"Bucket":"amzn-s3-demo-bucket","ObjectKey":"code/pkg.zip"}}' \
  --role-arn arn:aws:iam::123456789012:role/mwaa-pybash-test-role

レスポンスで新しいWorkflowVersionが返り、CodeSnapshottedAtはupdateした時刻に更新されました。list-workflow-versionsで見ると、2つのバージョンが並びます。

{
    "WorkflowVersions": [
        {
            "WorkflowVersion": "953984f3...",
            "IsLatestVersion": true,
            "CreatedAt": "2026-08-25T08:27:26.984000+00:00"
        },
        {
            "WorkflowVersion": "d495e078...",
            "IsLatestVersion": false,
            "CreatedAt": "2026-08-25T07:17:04.707084+00:00"
        }
    ]
}

新しい版にIsLatestVersion: trueが付いています。start-workflow-runで--workflow-versionを省略すると、このフラグがtrueの版が実行されます。実行するとv2になりました。VersionIdを省略したupdateは、その時点でS3にある最新オブジェクトを取り込みます。

list-workflow-versionsの各要素にはCodeが含まれません。どのWorkflowVersionがどのS3オブジェクトから作られたかは、APIからは後から追えませんでした。

古いコードで動かすには

単発で旧コードを動かすだけなら、start-workflow-runに--workflow-versionで旧版を指定します。

aws mwaa-serverless start-workflow-run \
  --workflow-arn arn:aws:airflow-serverless:ap-northeast-1:123456789012:workflow/pybash-snapshot-xxxxxxxxxx \
  --workflow-version d495e078...

結果はv1でした。

省略時の実行も旧コードにしたい場合は、--codeに古いVersionIdを明示してupdateします。S3の最新オブジェクトはv2のままです。

aws mwaa-serverless update-workflow \
  --workflow-arn arn:aws:airflow-serverless:ap-northeast-1:123456789012:workflow/pybash-snapshot-xxxxxxxxxx \
  --definition-s3-location '{"Bucket":"amzn-s3-demo-bucket","ObjectKey":"definitions/snapshot.yaml"}' \
  --code '{"S3Location":{"Bucket":"amzn-s3-demo-bucket","ObjectKey":"code/pkg.zip","VersionId":"G5eRYx6s..."}}' \
  --role-arn arn:aws:iam::123456789012:role/mwaa-pybash-test-role

3つ目のWorkflowVersionが作られ、get-workflowのCode.S3Locationに今度はVersionIdが入りました。

{
    "Code": {
        "S3Location": {
            "Bucket": "amzn-s3-demo-bucket",
            "ObjectKey": "code/pkg.zip",
            "VersionId": "G5eRYx6s..."
        }
    },
    "CodeSnapshottedAt": "2026-08-25T08:44:20.406374+00:00"
}

実行するとv1です。S3の最新はv2なのに、明示した版が取り込まれています。旧版が最新に戻ったわけではなく、旧コードを取り込んだ3つ目の版が作られ、それにIsLatestVersionが付いた形です。省略時の実行もv1になります。

S3からコードを消しても実行できるか

スナップショットが実行時にS3を参照しているのかを確かめるため、code/pkg.zipを全バージョン削除しました。

# 3つのVersionIdについて、それぞれ実行する
aws s3api delete-object \
  --bucket amzn-s3-demo-bucket \
  --key code/pkg.zip \
  --version-id "$VERSION_ID"

削除後はlist-object-versionsのVersionsとDeleteMarkersがどちらも空になります。head-objectは404、VersionIdを指定したget-objectはNoSuchVersionです。

この状態で3つのWorkflowVersionをそれぞれ--workflow-version指定で実行しました。

全部成功しました。少なくともこの3つのWorkflowVersionは、実行時にS3の元オブジェクトを参照していません。サービス側の管理領域に実行可能なスナップショットが保持されていると考えられます（内部の保存方式は公開されていません）。ただし、WorkflowVersionの保持期間や、削除後いつまで実行できるかは確認していません。元のオブジェクトはS3にも残しておくのが無難です。

バージョニングを無効にしたバケットでも同じことを試しました。create後に同じキーを上書きしてもv1のままで、オブジェクトを削除しても実行できました。S3のバージョニングはスナップショットの前提ではありません。一方で、特定の版を固定して取り込むにはVersionIdがいるので、その意味でバージョニングは必要です。ドキュメントが本番ではバージョニングを推奨しているのは、この意味だと理解しました。

定義を更新するときの2つの注意点

運用で注意がいると感じたのは、定義を修正するときの挙動でした。

--codeを省略すると、コードが引き継がれない

定義ファイルだけ直したいとき、update-workflowに--codeを付けずに呼ぶとどうなるか試しました。--definition-s3-locationと--role-arnだけで呼びます。

aws mwaa-serverless update-workflow \
  --workflow-arn arn:aws:airflow-serverless:ap-northeast-1:123456789012:workflow/pybash-nocode-xxxxxxxxxx \
  --definition-s3-location '{"Bucket":"amzn-s3-demo-bucket","ObjectKey":"definitions/nocode-update-v2def.yaml"}' \
  --role-arn arn:aws:iam::123456789012:role/mwaa-pybash-test-role

updateは成功します。しかしget-workflowを見ると、CodeとCodeSnapshottedAtがレスポンスから消えていました。実行すると、DAGの生成段階でNo module named 'mymod'になり、py_versionがFAILED、下流がUPSTREAM_FAILEDで終わります。

直前のWorkflowVersionを--workflow-versionで指定して実行するとv1で成功します。ワークフロー全体からコードが消えたのではなく、「新しく作られた版にコードがない」状態です。少なくともCodeについては、updateで直前の値は引き継がれません。定義だけ直す場合も、--codeは毎回付けます。

同じキーを再指定すると、そのときS3にあるコードが取り込まれる

では--codeを付ければ安全かというと、VersionIdを省略している限り、そのときS3にある最新オブジェクトが取り込まれます。

今回の検証では、別のワークフローで定義のtypoを直すためにupdateを2回かけていました。その間に、スナップショット検証のために同じキーcode/pkg.zipをv2で上書きしていたので、2回目のupdateでコードがv1からv2に変わっていました。定義しか触っていないつもりでも、コードが変わります。

整理するとこうなります。

• --codeを省略する → 新しい版にコードが引き継がれない
• --codeで同じキーだけを再指定する → そのときS3にある最新コードが取り込まれる

対策は、--codeに取り込みたいVersionIdを毎回明示すること、またはリリースごとに一意で上書きしないObjectKeyを使うことです。

複数のワークフローが同じコードのキーを共有している場合や、コードのアップロードと定義の修正が別の人・別のパイプラインで行われる場合は、特に引っかかりやすいところです。ドキュメント上は定義側の--definition-s3-locationにもVersionIdを指定できるので、再現可能なデプロイにしたいなら、定義とコードの両方を固定しておくのがよさそうです。

リリース記録に残すもの

APIからは「どのWorkflowVersionがどのコードか」を後から追えないので、update時に自分で記録しておきます。たとえば次のような対応表です。

update-workflowのレスポンスには入力したVersionIdが返らないので、コマンドの入力と一緒に残しておきます。

コードパッケージ固有の注意点

動かす中で引っかかった点が3つありました。

.shで終わるbash_commandはテンプレートファイル扱いになる

単一の.shをコードとしてアップロードし、bash_command: "bash run.sh"で呼ぶと、次のようなエラーになりました。

TemplateNotFound: 'bash run.sh' not found in search paths: '/usr/local/airflow/dags'

Airflowの仕様で、.shや.bashで終わるbash_commandはJinjaテンプレートファイルとして解決されます。従来のMWAAと同じ話ですが、コードとして.shを渡せるようになった分、踏む機会は増えそうです。

Airflow 2.8以降のPython DAGなら、literal()で囲んでレンダリング自体を無効化するのが今の書き方です。Airflow 3ではfrom airflow.sdk import literalでインポートできます。ただしMWAA Serverlessの定義はYAMLで、bash_commandには文字列しか書けないため、この方法は使えません。昔ながらの回避方法になりますが、末尾にスペースを入れて"bash run.sh "とすれば、文字列のまま実行されて成功しました。

BashOperatorの作業ディレクトリは、オペレーターの指定パスで変わる

ドキュメントには「BashOperatorのスクリプトは/usr/local/airflow/dagsを作業ディレクトリとして実行される」とあります。同じpwdをオペレーターの書き方だけ変えて実行すると、結果が違いました。

providersパスの方は、AirflowのBashOperatorのデフォルトどおり一時ディレクトリで動いています。旧パスairflow.operators.bash.BashOperatorの方だけdagsディレクトリになっていました。なぜ差が出るのかは分かりません。

bash run.sh のように「スクリプトを名前で呼ぶ」書き方は、どちらでも動きました。PATHに/usr/local/airflow/dagsが含まれていて、bashが引数のファイルをPATHから探すためです。相対パスでファイルを読み書きするスクリプトは、cwdを明示しておくのが確実です。

実は最初、bash_showでgrep -n 'VERSION =' mymod.pyも出力させていました。providersパスで書いていたので、このgrepはNo such file or directoryで失敗していました。XComの値だけを見て検証を進めてしまい、ログを見返して気付いた次第です。

同梱したパッケージより、プリインストール版が優先される

ドキュメントに「プリインストール済みのパッケージを別バージョンで同梱しても、プリインストール版が使われる」とあります。lz4（プリインストールは4.4.4）を4.3.3で同梱して確認しました。

importlib.metadata.version("lz4") → 4.3.3
lz4.__version__                   → 4.4.4
lz4.__file__                      → /opt/mwaa-serverless-venv/lib64/python3.12/site-packages/lz4/__init__.py

importlib.metadata.version()とpip showは同梱した4.3.3を示しました。しかしimportされたモジュールは、__file__と__version__のとおりプリインストール済みの4.4.4です。ドキュメントどおりですが、配布メタデータと実際のimport元が食い違います。実行時にバージョンを確認するなら、__file__やパッケージ自身のバージョン属性も見ておきましょう。

pip listで見ると、プリインストールされているパッケージはドキュメントの表より多く、tabulateやrequests、redshift-connectorなども入っていました。同梱する前に確認しておくと、無駄なzipの肥大化を避けられます。

そのほか気付いたこと

• python_callableの戻り値はXComに載り、get-task-instanceのXcom.return_valueにJSON文字列として入る。BashOperatorも最終行がreturn_valueになる
• op_kwargsはYAMLから渡せる。paramsと{{ params.xxx }}の組み合わせも動き、start-workflow-run --override-parametersで上書きできる。一方、--override-parametersの値はタスクの環境変数には入っていなかった
• {{ ds }}は使えた。一方{{ run_id }}をbash_commandに書くと、DAG template rendering failed: 'run_id' is undefinedとなり実行前に失敗した。以前の記事でti.run_idが参照できなかったのと同じく、テンプレートで使える変数は制限されている
• VPCなしの構成では、boto3からS3には届いたがsts get-caller-identityはConnectTimeoutErrorになった
• BranchPythonOperatorはis not supportedでcreate時に弾かれる。PythonVirtualenvOperatorとPythonSensorは、python_callableがPythonOperator専用のパラメーターだとして弾かれる。カスタムコード向けに使えるのはPythonOperatorとBashOperatorの2つ

まとめ

MWAA ServerlessでPythonOperatorとBashOperatorが使えるようになりました。コードはS3に置いて--codeで渡します。

スナップショットに関する挙動を操作ごとに整理すると、次のとおりです。

S3から元オブジェクトを消しても、確認した3つのWorkflowVersionは実行できました。一方でget-workflowのVersionIdは明示したときしか返らず、list-workflow-versionsにはCodeが含まれません。どのWorkflowVersionがどのコードかは、S3のVersionIdとzipのハッシュを自分で記録しておくことになります。

コードを渡せるようになったこと自体より、ワークフローのバージョンとコードの対応をどう記録するかの方が、運用では重要になりそうです。

どなたかの参考になりましたら幸いです。