[アップデート] MWAA Serverlessにオペレーターが19個追加されワークフローから別のワークフローを起動できるようになりました

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

Amazon SageMaker Unified Studio Workflowsで、新しいオペレーターがサポートされました。対象はAmazon Bedrock、Amazon S3 Tables、Amazon S3 Vectors、AWS Glue Data Catalogです。

https://aws.amazon.com/about-aws/whats-new/2026/07/apache-airflow-operators-amazon-sagemaker-unified-studio-workflows/

これらのオペレーターはMWAA Serverlessで使えます。Unified Studioのビジュアルクリエイターからでも、YAML定義からでも指定できます。

追加分の中にワークフローを起動するオペレーターが含まれていたので、実際に動かして確認しました。

追加された19個の内訳

追加先はMWAA Serverlessのサポートオペレーター一覧です。

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

告知には「19 new operators」とありますが、どのオペレーターかは列挙されていません。告知が挙げた5サービスはBedrock guardrail、S3 Tables、S3 Vectors、Glue Data Catalog、MWAA Serverlessです。この一覧のうち該当するものを数えると19個になります。

合計19個です。AWSが「この19個」と明示したわけではなく、カテゴリと数の一致からの割り出しです。

BedrockはBedrockInvokeModelOperatorやBedrockCreateKnowledgeBaseOperatorなどが以前からありました。今回増えたのはguardrail関連の3個です。告知本文も「guardrailsの管理」と書いています。

これらのオペレーターは、いずれもApache Airflowのオープンソース版Amazonプロバイダーに実装済みです。

https://airflow.apache.org/docs/apache-airflow-providers-amazon/stable/operators/index.html

今回のアップデートで、MWAA Serverlessから呼べるようになりました。

ワークフローから別のワークフローを起動する

追加分のうち、Workflowsカテゴリの2つを試しました。

• MwaaServerlessCreateWorkflowOperator - ワークフローを作成する
• MwaaServerlessStartWorkflowRunOperator - ワークフロー実行を開始する

MwaaServerlessStartWorkflowRunOperatorは、ワークフローの中から別のワークフローを起動するオペレーターです。AirflowでいうTriggerDagRunOperatorのような使い方が、MWAA Serverlessでもできます。

以前、MWAA ServerlessのEventBridge対応を検証したとき、ワークフローの連鎖について次のように書いていました。

https://dev.classmethod.jp/articles/update-amazon-mwaa-serverless-eventbridge-events/

MWAA Serverlessには他のワークフローの完了を待つ手段が乏しく、センサーで待つとその間のタスク実行時間が課金されます。Succeededイベント→Lambda→StartWorkflowRunという連携が組めるのは個人的にうれしいポイントです。

このときは連鎖にEventBridgeとLambdaを挟んでいました。MwaaServerlessStartWorkflowRunOperatorがあれば、上流ワークフローの末尾にタスクを置くだけで下流を起動できます。実際に動くか確認しました。

検証の構成

ワークフローを2つ用意しました。

• 下流ワークフロー: S3にマーカーファイルを1つ書き込む（S3CreateObjectOperator）
• 上流ワークフロー: S3にマーカーを書いたあと、MwaaServerlessStartWorkflowRunOperatorで下流を起動する

下流のYAML定義はこうです。

downstream_wf:
  dag_id: downstream_wf
  schedule: null
  default_args:
    owner: airflow
    start_date: "2024-01-01"
  tasks:
    write_marker:
      operator: airflow.providers.amazon.aws.operators.s3.S3CreateObjectOperator
      task_id: write_marker
      s3_bucket: amzn-s3-demo-bucket
      s3_key: markers/downstream_ran.txt
      data: "downstream executed"
      replace: true

上流のYAML定義です。trigger_downstreamタスクで下流ワークフローのARNを指定します。

upstream_wf:
  dag_id: upstream_wf
  schedule: null
  default_args:
    owner: airflow
    start_date: "2024-01-01"
  tasks:
    write_upstream_marker:
      operator: airflow.providers.amazon.aws.operators.s3.S3CreateObjectOperator
      task_id: write_upstream_marker
      s3_bucket: amzn-s3-demo-bucket
      s3_key: markers/upstream_ran.txt
      data: "upstream executed"
      replace: true
    trigger_downstream:
      operator: airflow.providers.amazon.aws.operators.mwaa_serverless.MwaaServerlessStartWorkflowRunOperator
      task_id: trigger_downstream
      workflow_arn: arn:aws:airflow-serverless:ap-northeast-1:123456789012:workflow/downstream-wf-xxxxxxxxxx
      dependencies:
        - write_upstream_marker

MwaaServerlessStartWorkflowRunOperatorの必須パラメーターはworkflow_arnです。ほかにoverride_parametersとworkflow_versionを指定できます。

実行ロール

MWAA Serverlessはワークフローごとに実行ロールを1つ持ちます。信頼ポリシーにはairflow-serverless.amazonaws.comを指定します。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "airflow-serverless.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

上流の実行ロールには、S3書き込みとCloudWatch Logsに加えてairflow-serverless:StartWorkflowRunを付けました。下流を起動するのはこの権限です。

{
  "Effect": "Allow",
  "Action": ["airflow-serverless:StartWorkflowRun"],
  "Resource": "*"
}

動かした結果

ワークフローはYAMLをS3に置いてからcreate-workflowで作成します。上流のYAMLに書くworkflow_arnは下流を作らないと決まらないので、下流から先に作ります。

まず下流を作成します。

aws mwaa-serverless create-workflow \
  --name downstream-wf \
  --definition-s3-location '{"Bucket":"amzn-s3-demo-bucket","ObjectKey":"definitions/downstream.yaml"}' \
  --role-arn arn:aws:iam::123456789012:role/mwaa-sls-exec

レスポンスで下流のARNが返ります。

{
    "WorkflowArn": "arn:aws:airflow-serverless:ap-northeast-1:123456789012:workflow/downstream-wf-xxxxxxxxxx"
}

このARNを、先ほどの上流YAMLのtrigger_downstreamタスクのworkflow_arnに記入します。書き換えたら上流を作成します。

aws mwaa-serverless create-workflow \
  --name upstream-wf \
  --definition-s3-location '{"Bucket":"amzn-s3-demo-bucket","ObjectKey":"definitions/upstream.yaml"}' \
  --role-arn arn:aws:iam::123456789012:role/mwaa-sls-exec

あとは上流を実行します。

aws mwaa-serverless start-workflow-run \
  --workflow-arn arn:aws:airflow-serverless:ap-northeast-1:123456789012:workflow/upstream-wf-xxxxxxxxxx

上流の実行はSUCCESSで終わり、write_upstream_markerとtrigger_downstreamの2タスクが完了しました。下流側を見ると、上流の実行時刻のあとに新しい実行が1件増えていて、状態はSUCCESS、RunTypeはON_DEMANDでした。下流のマーカーファイルも書き込まれています。上流のtrigger_downstreamタスクが下流を起動しました。

起動された下流の実行は、コンソールやCLIからの手動実行と同じON_DEMANDとして記録されます。オペレーター経由の起動を区別するフィールドは見当たりませんでした。

権限を外すとどうなるか

このオペレーターに必要な権限は、ドキュメントに載っていません。上流の実行ロールからairflow-serverless:StartWorkflowRunだけを外して、もう一度実行してみました。

結果はFAILEDです。タスク単位で見ると、S3CreateObjectOperatorのタスクはSUCCESSでした。MwaaServerlessStartWorkflowRunOperatorのタスクだけがFAILEDです。下流側には新しい実行が増えておらず、下流のマーカーも書かれていません。

{
  "TaskInstanceId": "..._write_upstream_marker_1",
  "Status": "SUCCESS",
  "OperatorName": "S3CreateObjectOperator"
},
{
  "TaskInstanceId": "..._trigger_downstream_1",
  "Status": "FAILED",
  "OperatorName": "MwaaServerlessStartWorkflowRunOperator"
}

list-task-instancesはタスクごとにOperatorNameとStatusを返すので、どのオペレーターで落ちたかは追えます。ただし失敗の理由はここには出ないため、原因はタスクログで確認します。

完了待ちまではできない

起動はできるようになりましたが、実行の完了を待つMwaaServerlessWorkflowRunSensorはサポート一覧にありません。MWAA Serverlessは一覧のオペレーターだけを受け付けるので、下流を起動しても、その完了を待って次のタスクに進むことは今のところできません。先ほど引用した「他のワークフローの完了を待つ手段が乏しい」状況は、起動の手が増えた一方で、完了待ちは残ったままです。

完了を検知したいなら、これまでどおりEventBridgeへ配信されるイベントを使うことになります。

まとめ

MWAA Serverlessで使えるオペレーターが19個増えました。

• 追加先はMWAA Serverlessのサポートオペレーター一覧。Unified Studioを使わずYAML定義からも利用できる
• 内訳はMWAA Serverless 2個、Glue Data Catalog 4個、Bedrock guardrail 3個、S3 Tables 6個、S3 Vectors 4個
• MwaaServerlessStartWorkflowRunOperatorでワークフローから別のワークフローを起動できる。上流から下流を起動する構成が動いた。EventBridgeとLambdaを挟まずに連鎖を組める
• 下流の起動には、上流の実行ロールにairflow-serverless:StartWorkflowRunが必要。外すと起動タスクが失敗した
• 完了を待つMwaaServerlessWorkflowRunSensorはサポート一覧になく、完了待ちは引き続き課題になりそう

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