本記事は、Accenture の AABG Global Innovation Lead である Amanda Jensen 氏、および Accenture の AABG Lead GenAI Architect である Mike Earley 氏による、AWS との協業に基づくゲスト投稿です。 エグゼクティブサマリーと要点 ある大手企業は、Accenture と協業し、AWS 上にエージェント型 AI（agentic AI）ソリューションを展開することで、Direct Ship（直送）請求業務の変革を実現しました。このソリューションは、手作業による例外処理を、リアルタイムで AI 主導のインサイトとガイダンスに置き換えるものです。その結果、展開からわずか 5 週間で、業務上の摩擦を軽減し、課題解決を加速し、売上の実現（revenue realization）を向上させる、迅速かつビジネス主導の変革がもたらされました。要点は以下のとおりです。 ビジネス成果に整合し、スケーラブルなクラウドインフラストラクチャによって支えられたとき、AI は数か月で測定可能なインパクトをもたらすことができます。これは、企業の再創造（reinvention）に向けた実践的な道筋を示すものです。 エージェント型 AI は、業務を事後対応型の例外管理から、プロアクティブでインサイト主導の意思決定へと転換し、価値の高い案件のより迅速な解決を可能にします。 ナレッジの一元化と対話型インターフェイスは、生産性を高め、専門知識をスケールさせ、手作業のプロセスへの依存を減らして、経験豊富なスタッフの負担を軽減します。 はじめに 大量のトランザクションを扱う企業にとって、エンタープライズリソースプランニング（ERP）システムにおける手作業の例外処理は、重要なビジネスプロセスを遅延させ、最終的にはお客様満足度と収益認識（revenue recognition）に影響を及ぼすボトルネックを生み出します。こうした複雑なビジネスプロセス業務を大規模に管理することは、重大な運用上の課題をもたらします。ERP の例外管理に対する従来型のアプローチでは、絶えず増大するビジネス要求に追いつくことができず、企業には、AI エージェントが自律的に推論し、意思決定を行い、行動できるような、AI 主導のアプローチによって卓越した業務運営を実現することが求められています。本記事では、Accenture が AWS のサービスを用いて、ERP のビジネスプロセス例外管理を変革するエージェント型 AI ソリューションをどのように構築したかをご紹介します。 ERP 業務における課題 Direct Ship（直送）業務は、大企業にとって特有の複雑さを伴います。成熟した ERP ランドスケープにおいてさえ、注文量、プロセス構成の深さ、そして例外シナリオの数の多さから、運用チームが請求上の問題を効率的に特定し、トリアージし、解決することは困難です。 ワークフローが ERP を越えて、インシデント管理、物流、コンプライアンスの各システムにまで広がると、課題はさらに複雑になります。これらの領域では、インターフェイスの障害、データの不整合、引き継ぎ時のギャップが頻繁に発生します。プロセスに関するナレッジは経験豊富なチームメンバーに集中しがちであり、標準作業手順書（SOP）が複数のドキュメントにまたがることも多く、オンボーディングの難しさや解決までの時間の長さを招いています。 海外向けの注文は、通関書類、国ごとの税および関税のルール、付加価値税（VAT）コンプライアンス、通貨の取り扱い、輸出規制などにより、さらに複雑さを増します。その結果として、請求の遅延、収益のギャップ、コンプライアンスリスクが生じます。 エージェント型 AI ソリューション これらの課題に対処するため、Accenture は Amazon Bedrock や Bedrock AgentCore といった AWS サービスを用いてエージェント型 AI ソリューションを開発し、ERP の例外管理を、手作業で労力を要する業務から、自律的でエージェント主導の解決へと転換しました。このソリューションは、オペレーターの体験を変革するために連携して機能する 2 つの主要コンポーネント、すなわち Digital Assistant と、エンタープライズシステムとの統合機能で構成されています。Digital Assistant は、SOP、ジョブエイド、組織内のナレッジを単一のアクセス可能なシステムに一元化する対話型 AI インターフェイスとして機能し、チームメンバーが自然言語で質問し、複雑なシナリオを解決するための文脈に沿ったガイダンスを受け取れるようにします。これは、Direct Ship プロセスに関する深いナレッジを備えた AI 搭載のアシスタントであり、解決策を推奨し、SAP データへのリアルタイムアクセスを提供することで、複数のシステムを行き来する必要性を排除します。エージェントは文脈を処理し、会話履歴を保持し、実行可能なインサイトを提供します。このソリューションは、SAP S/4HANA や認証用の Okta を含む既存のエンタープライズシステムと統合されており、AI エージェントがエンタープライズのセキュリティ基準を維持しながら、リアルタイムの運用データにアクセスできるようになっています。 ソリューションアーキテクチャ 本ソリューションは、6 つの AWS サービスを組み合わせており、それぞれがエンタープライズのセキュリティとスケーラビリティを備えた AI 機能を提供するうえで特定の役割を担うように選定されています。 以下では、これらのサービスが実際にどのように連携して機能するかを説明します。（1）チームメンバーが、請求の例外について Digital Assistant に自然言語で質問します。（2）Strands Agents SDK がリクエストを評価し、どのツールを呼び出すかを判断して、解決ワークフローをオーケストレーションします。（3）AWS Lambda が SAP S/4HANA に安全に接続し、リアルタイムの注文、請求、出荷のデータを取得します。（4）Amazon RDS が、事前にロードされた例外レポートへの高速なアクセスを提供し、エージェントがパターンを相互参照して根本原因を特定できるようにします。（5）Amazon Bedrock AgentCore が、複数ターンにわたるやり取りの会話コンテキストを保持することで、エージェントが以前の質問を記憶します。（6）エージェントがすべてのデータソースを統合し、具体的な解決手順とともに優先順位付けされた推奨事項を返します。 Amazon Bedrock ：単一の API を通じて、主要な AI 企業が提供する高性能な基盤モデルへのアクセスを提供します。本ソリューションでは、Bedrock が Digital Assistant の能力を支え、複雑な請求シナリオの理解、SOP の解釈、文脈に沿った推奨事項の生成を可能にします。 Strands Agents SDK ：エージェントを LLM、ツール、およびネイティブな AWS サービスとシームレスに統合し、基盤モデルの推論能力を用いて、入力の評価、次のステップの判断、ツールの自律的な選択を可能にします。これにより、ハードコードされた意思決定ツリーが不要になり、エージェントが新しいタイプの例外にも対応できるようになります。 Amazon Bedrock AgentCore ：複雑なメモリインフラストラクチャなしに、短期的なワーキングメモリのコンテキストを管理します。これにより、エージェントがセッションの前半で議論した内容を記憶する、自然な複数ターンの会話が可能になります。gateway、runtime、observability といった AgentCore の追加機能により、本ソリューションは将来的に数百の AI エージェントへとスケールできます。 AWS Lambda ：SAP S/4HANA を含むエンタープライズシステムに安全にアクセスするための、サーバーレスなゲートウェイとして機能します。Lambda のイベント駆動型アーキテクチャにより、アイドル時のコストが発生しない一方で、エージェントのクエリに低レイテンシーで応答できます。 Amazon RDS ：事前処理された SAP S/4HANA のレポートデータを保存し、例外の特定やステータス更新のための高速なクエリをサポートします。この分離により、SAP システムの負荷を低く保ちながら、エージェントに運用データへのサブ秒単位のアクセスを提供します。 AWS Step Functions ：データソースからのレポートデータのロードワークフローをオーケストレーションし、エージェントのクエリに対してデータが最新かつ利用可能な状態であることを保証します。組み込みのリトライロジックと視覚的なワークフロー監視により、カスタムのオーケストレーションコードなしで信頼性を提供します。 メリットとビジネス成果 Accenture は AWS との協業により、5 週間で稼働するソリューションを提供し、テクノロジーと SAP のビジネスプロセスにおける実証済みの専門性を示しました。このソリューションは、反復的な管理作業を最小化し、例外管理における手作業の労力を削減しました。エージェントが SAP データへの即時アクセスを提供し、複数のシステムを行き来して切り替える必要性を排除したためです。推奨されるアクションにより、運用チームは請求の例外を事後対応ではなくプロアクティブに処理できるようになり、価値の高い注文が即座に対応され、収益の遅延が削減されます。新しいチームメンバーは、対話的なやり取りを通じて組織内のナレッジに即座にアクセスでき、Digital Assistant が定型的な質問における経験豊富なスタッフへの依存を減らすことで、シニアなチームメンバーが複雑なシナリオに集中できるようになります。SOP やトラブルシューティングガイドは、一元化されたナレッジベースを通じて常に最新かつアクセス可能な状態に保たれ、組織全体で一貫したプロセス実行を維持するのに役立ちます。 企業は、Accenture が AWS サービスを用いて構築したエージェント型 AI ソリューションを通じて、自社のビジネスプロセス業務を変革できます。このソリューションは、明確な目標を、クラウドネイティブな AI 機能や、実証済みの ERP ドメイン専門性を持つパートナーと組み合わせたとき、AI 主導のプロセス変革が、数年ではなく数週間で測定可能なビジネス価値をもたらせることを実証しています。企業が ERP 業務の簡素化と自動化を進め続ける中で、エージェント型 AI は、運用チームを強化しビジネス成果を推進するための効果的なアプローチとなります。今後を見据えると、同じアーキテクチャパターンは、調達、在庫管理、その他の SAP モジュールにも拡張できます。 AWS SAP スペシャリストチーム または Accenture AWS Business Group にお問い合わせいただき、SAP 向けのエージェント型 AI ワークショップを予約して、ERP 管理のモダナイゼーションとビジネス成果の向上に向けた方法をぜひご検討ください。また、 AWS GitHub リポジトリ にある、すべての Bedrock AgentCore サービス向けのすぐに使えるコードサンプルを活用して、開発を素早く開始することもできます。 Accenture AWS パートナースポットライト Accenture は、AWS への移行と AWS 上での運用管理に向けたエンドツーエンドのソリューションを提供する、AWS Premier Tier Services Partner および MSP です。Accenture と AWS の戦略的な協業体制である Accenture AWS Business Group（AABG）と協働することで、差別化された製品やサービスを提供するためのイノベーションのペースを加速できます。 AWS SAP スペシャリストチームにお問い合わせ | Accenture にお問い合わせ <!-- '"` --> 本ブログの翻訳は Amazon Quick による自動翻訳を行い、パートナー SA 松本がレビューしました。原文は こちら です。

はじめに こんにちは、クラウドセキュリティグループの小林です。 普段は AWS を中心としたクラウド環境のセキュリティ運用と改善を担当しています。 本記事では、GuardDuty のアラートを全件目視で確認していた運用を、AWS Security Incident Response（以下 SIR）の導入と自作の Slack 連携でどのように変えたかを紹介します。 本記事の対象読者 GuardDuty のアラート対応に負担を感じている方 AWS Security Incident Response の導入を検討している方 インシデント対応の導線を Slack に統合したい方 背景 当社はマルチアカウントの AWS 環境を GuardDuty で監視しています。 SIR 導入前の運用は、アラートが発火するたびにマネジメントコンソールを開き、GuardDuty の検出内容・関連リソース・CloudTrail ログなどを確認し、実害のある検知なのか過検知なのかを毎回判断する運用でした。 アラート 1 件ごとの作業は短時間でも、アカウント数に比例して検出件数が増えるため、確認に使う時間は無視できない量になっていました。 特に 2026 年 5 月〜6 月、GitHub Actions に起因する検知アラートが目立つようになりました。 GitHub がホストするランナーの IP アドレスは多数の利用者で共有されており、他の利用者による悪用などを原因としてこの IP が GuardDuty の参照する脅威インテリジェンスに登録されると、同じ IP 帯から実行される正規の GitHub Actions の通信まで脅威由来として検知され、アラートが多発します。 このアラートへの対応が、SIR の導入を検討したきっかけになりました。 AWS Security Incident Response とは？ AWS Security Incident Response は、セキュリティイベントの検知から対応までを支援する AWS のマネージドサービスです。 SIR は GuardDuty の検出結果と、Security Hub 経由で連携したサードパーティー製品の検出結果を取り込み、自動でトリアージします。 過検知と判定されたものはアーカイブされ、緊急度の高いものだけがケースとして起票されます。 これにより、それまで人手で行っていた確認とアーカイブの作業の大部分をサービスに置き換えられます。 なお、一部の Finding タイプは自動トリアージの対象外です。 ケース化された検出結果は、分析から封じ込め、クローズまでのライフサイクルがケース上に記録され、対応状況を一元的に追跡できます。 また、AWS サポートが必要なケースについては、SIR エンジニア（セキュリティ脅威のトリアージ・調査・封じ込めを支援する AWS の専門エンジニア）から 15 分以内に初回応答があり、ケースクローズまで継続して調査の支援を受けられます。 例えば、アカウント乗っ取りのような明確なインシデントだけでなく、疑わしい検知が本物の侵害かどうかの調査や、GuardDuty のトリアージ設定・抑制ルールに関する問い合わせなども AWS サポートが必要なケースとして起票できます。 費用面では、エンタープライズサポート契約があれば SIR のサービス全体を追加費用なしで利用できます。 Slack 連携 当社はエンタープライズサポートを契約していたため、SIR は追加コストなしですぐに導入できました。 しかし、運用に載せると導線の問題が見えてきました。 SIR ケースの確認や操作にはマネジメントコンソールへのログインが必要なため、社内標準のコミュニケーションツールである Slack での会話とマネジメントコンソール上の記録を行き来する手間があり、ケースの変化に気づくのが遅れる懸念もありました。 そこで AWS 公式のサンプル実装（ sample-aws-security-incident-response-integrations ）を参考に、Slack との双方向連携を作成しました。 起票は Slack のスラッシュコマンドから行います。 /sir create を実行すると入力モーダルが開き、ケースタイプやタイトル、影響を受けるアカウントなどを入力してケースを作成できます。 説明欄には「何が起きたか」「現在の影響」などの定型見出しが初期表示されるため、埋めるだけで報告の体裁が整います。 起票は誰でもできる一方、ステータス変更や内容の編集といった影響の大きい操作は管理者に限定しており、変更時、誰が何のアクションを実行したか Slack 上に記録するようにしています。 起点が Slack か SIR かを問わず、ケースが作成されるとチャンネルに初報が投稿され、指定した個人やグループへメンションが送られます。 以後の更新やコメントの通知はすべて初報へのスレッド返信として集約しています。 対応中はスレッド内の発言や画像が SIR ケースに自動で記録され、SIR 側のコメントもスレッドに届くため、スレッドがそのまま作業記録になり、Slack から SIR へ転記する手間と記録漏れがなくなります。 複数のケースが動いているときは /sir list でオープン中のケースを一覧でき、一覧の各ケースに対して内容の編集やステータス変更も行えます。 ケースは対応完了まで Open のまま残り、クローズ時にもスレッドへ通知されます。 アーキテクチャと設計上の注意点 この連携は API Gateway、Lambda（4 関数）、EventBridge カスタムバス、DynamoDB に、Slack の認証情報などコードに含めたくない設定を保管する Secrets Manager を加えたサーバーレス構成で、Terraform で管理しています。 主要な流れは次の図のとおりです。 SIR のイベント通知機能はメールのみで、ケースの変化をイベントとして受け取る手段がありません。 そのため SIR から Slack への方向は、poller が SIR の API を定期的に呼び、DynamoDB に保存した前回のスナップショットと比較して差分を検知するポーリング構成になります。 ポーリング間隔は、未クローズのケースがある間は 1 分、なければ 5 分になるよう、poller が自身をトリガーする EventBridge ルールを実行のたびに書き換えて調整しています。 対応中の即時性と平常時の API 呼び出し削減を両立するためですが、それでも通知には最大 1〜5 分の遅延が残ります。 逆方向の Slack から SIR への経路では、コマンドやボタン操作への応答を 3 秒以内に返すという Slack の制約が存在するため、入口では即座に応答だけを返し、実処理は後段の Lambda へ非同期で委譲する構成となっています。 ただし、モーダルの入力エラーのように応答の内容そのものを返す場面は非同期にできないため、API Gateway のルートは同期・非同期の 2 本に分けています。 また、モーダルを開くための trigger_id も 3 秒で失効するため、EventBridge で 5 分ごとに Lambda を空起動してコールドスタートを避けています。 なお、リクエストの真正性は入口での Slack 署名の検証で確認しています。 双方向に同期させると、Slack からの変更を poller が SIR 側の変化として検知し、Slack へ通知し返すループが起こり得ます。 これは、Slack 起点の変更時に DynamoDB へ書き込むケース単位のフラグを poller が確認して通知をスキップする仕組みと、SIR へ書き込むコメントにタグを付けて同期対象から除外する仕組みで防いでいます。 導入後の効果 導入後は SIR の自動トリアージが機能し、多い月では過検知の 70% が自動アーカイブされていました。 その大半は、背景で挙げた GitHub Actions 起因の検知です。 導入のきっかけになったアラートを、人手を介さずに処理できるようになりました。 人が確認するのは、緊急度が高いと判定されてケース化されたアラートだけになり、その対応は Slack のスレッドの中で完結するため、対応漏れも起こりにくくなりました。 残っている課題と今後の対応 現時点で見えている課題は 2 点あります。 SIR の自動トリアージの対象外となる Finding タイプがあることです。対象外の検知は SIR を経由しないため、これまでどおり人が確認する必要があります。 自動アーカイブした履歴が SIR 側に残らないことです。何がいつ自動アーカイブされたのかを確認するには、CloudTrail のログを調べるしかないのが現状です。 今後は、この自動アーカイブ履歴を可視化する仕組みづくりに取り組む予定です。 SIR が過検知と判定したものでも社内の基準では調査が必要になるケースが考えられるため、SIR の判定を後から確認できる状態にしておきたいと考えています。 まとめ GuardDuty のアラートを全件目視する運用は、件数が増えると維持が難しくなります。そこで当社では、過検知の仕分けを SIR に任せて人が判断する対象を絞り、残ったケースの操作と通知を Slack に集約することで、本当に対応が必要なアラートに集中できる体制にしました。 エンタープライズサポート契約があれば SIR は追加費用なしで有効化できるので、まずは自動トリアージがどの程度の過検知を仕分けられるかを確かめてみてください。そのうえでマネジメントコンソール中心の導線に不便を感じたら、公式サンプルを土台に Slack 連携を検討することをおすすめします。 なお、アラート分析そのものについても、SOC AI Agent による自動化で、分析の高速化・属人化の解消・担当者の負荷軽減を進めています。詳細は「 アラート疲弊からの脱却へ ― SOC業務におけるAIエージェント活用の実践 」をご覧ください。 本記事が、同様の課題を抱えている方の参考になれば幸いです。 参考資料 AWS Security Incident Response ドキュメント sample-aws-security-incident-response-integrations (aws-samples)

本記事は 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 &gt; trust-policy.json &lt;&lt; '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 &amp;&amp; zip -r ../code_bundle.zip . &amp;&amp; cd .. # Upload to S3 aws s3 cp code_bundle.zip s3://$BUCKET/code/code_bundle.zip --region $REGION テスト用のサンプル CSV ファイルをアップロードします。 cat &gt; sales_data.csv &lt;&lt; '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::&lt;your-account-id&gt;: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 の料金ページ を参照してください。フィードバックをお待ちしています。 著者について Pradeep Kumar Nalluri AWS のソフトウェア開発エンジニアで、スケーラブルなアプリケーションの設計と開発を専門としています。休日はテレビ番組や映画を観て過ごしています。 Karthik Seshadri AWS のシニアソフトウェア開発エンジニアで、ビッグデータ技術のオーケストレーションを専門としています。サーバーレス技術、データエンジニアリング、スケーラブルなサービスの構築に情熱を注いでいます。仕事以外では、旅行やさまざまなスポーツを楽しんでいます。 Aritra Ghosh Amazon Web Services (AWS) のシニアプロダクトマネージャーで、Amazon Managed Workflows for Apache Airflow (Amazon MWAA) と Amazon SageMaker Unified Studio の製品開発を率いています。仕事以外では、スカッシュとジム通いを楽しんでいます。 Sriram Ramarathnam AWS Analytics で AWS Glue、AWS Data Pipeline、Managed Serverless Airflow を担当するソフトウェア開発マネージャーです。チームでは、サーバーレスとプロビジョンド両方のコンピューティング提供形態にまたがるオーケストレーション領域の難しい課題に取り組んでいます。