トップ > AWS Batch > Step Functions + ECS RunTaskで頭打ちになったバッチ処理基盤をAWS Batchへ移行

Step Functions + ECS RunTaskで頭打ちになったバッチ処理基盤をAWS Batchへ移行

AWS Batch Amazon Elastic Container Service AWS Step Functions
記事タイトルとURLをコピーする

はじめに

こんにちは、アプリケーションサービス本部の遠藤です。

弊社で開発・運用を担当しているあるシステムには、大量のデータを定期的に処理するバッチ処理機能があります。処理対象は複数のグループに分かれており、グループ内の処理は決まった順序で（直列で）行う必要がある、という要件がありました。

このバッチ処理基盤は元々AWS Step Functions + Amazon ECS（Fargate）で構築していましたが、処理規模の拡大に伴い複数のAWSサービスクォータに頭打ちになり、最終的にAWS Batchへ移行しました。本記事では、その検討の過程で気になった点を公式ドキュメントや実測で調査した内容と、実際にどう設計を変えたかを紹介します。

元々の構成

実行基盤にはAWS Step Functionsを採用していました。理由は以下の通りです。

  • 1回の処理は複数の長時間タスクで構成されており、AWS Lambdaの実行時間制限には適さない
  • タスクによっては並列実行したいもの、順番を厳密に守る必要があるものが混在する
  • タスクの合間に人の判断（承認等）が入ることもある
  • 要件の解像度が上がるにつれて、タスクの実行順序や構成を柔軟に組み替えたい

これらを踏まえ、並列実行・順序制御・待機を組み合わせられるStep Functionsを選定しました。

処理対象のグループ一覧をDistributed Mapで回し、各グループに対してさらにInline Map（MaxConcurrency: 1）でグループ内の項目一覧を回し、項目ごとにECS RunTask.syncでコンテナを起動する、という構成です。1コンテナ＝1項目とし、グループ内の処理順序はInline MapのMaxConcurrency: 1で担保していました（MaxConcurrencyは本来「並列数の上限」を指定するパラメータですが、上限を1にすることで疑似的に直列実行を実現する、よく使われる小技です）。

flowchart TD
    Start(["グループ一覧"]) --> DM

    subgraph DM["Distributed Map（グループごとに並列実行）"]
        direction LR

        subgraph IM1["グループ1: Inline Map（MaxConcurrency = 1）"]
            direction TB
            A1["ECS RunTask.sync<br/>1コンテナ = 項目1"] --> A2["ECS RunTask.sync<br/>1コンテナ = 項目2"] --> A3["ECS RunTask.sync<br/>1コンテナ = 項目N"]
        end

        subgraph IM2["グループ2: Inline Map（MaxConcurrency = 1）"]
            direction TB
            B1["ECS RunTask.sync<br/>1コンテナ = 項目1"] --> B2["ECS RunTask.sync<br/>1コンテナ = 項目2"] --> B3["ECS RunTask.sync<br/>1コンテナ = 項目N"]
        end

        subgraph IMN["グループN: Inline Map（MaxConcurrency = 1）"]
            direction TB
            C1["ECS RunTask.sync<br/>1コンテナ = 項目1"] --> C2["ECS RunTask.sync<br/>1コンテナ = 項目2"] --> C3["ECS RunTask.sync<br/>1コンテナ = 項目N"]
        end

        %% グループ自体を横並びに強制するための不可視リンク
        IM1 ~~~ IM2 ~~~ IMN
    end

グループ単位（Distributed Map）は並列に実行されますが、グループの中（Inline Map）はMaxConcurrency: 1によって1件ずつ順番に実行され、かつ1コンテナが1項目しか処理しないため、項目数が多いグループほどコンテナの起動回数も比例して増える構成になっていました。

最初の壁: AWSサービスクォータとステート間のペイロード制限

処理対象の規模が大きくなるにつれ、2つの壁にぶつかりました。

1つ目は、Step Functionsのステート間で受け渡せるデータサイズの上限（256KB）です。ステート間で大量のデータをやり取りする必要があったため、ステート間の受け渡しをAmazon S3経由に変更しました。

2つ目は、各種AWSサービスのクォータです。当時整理した制限値の一部を紹介します。

サービス 制限項目 制限値 引き上げ可否
AWS Fargate タスク起動レート バースト100タスク／継続20タスク/秒 可能
Amazon ECS RunTask APIレート 20リクエスト/秒（1リクエストで最大10タスク起動可） 可能
Step Functions（Standard） StateTransitionスロットリング 800/秒 可能
Step Functions Distributed Map並列子実行数 10,000 不可

これらの引き上げ申請や、Lambda・ECSタスクの並列上限数の調整で、しばらくは安定稼働させることができました。

根本的な壁: 並列数の増加によるAPIスロットリング

しかし、処理対象のグループ数が増えて並列数が増加するにつれて、ECS RunTaskのAPIスロットリングが発生するリスクが無視できなくなりました。並列数に比例してRunTask呼び出し回数が増える構成だったため、クォータ引き上げだけでは根本解決にならないと判断し、アーキテクチャの見直しに着手しました。

候補に挙がったのがAWS Batchです。AWS Batchは内部キューで段階的にジョブをスケジューリングするため、呼び出し側が大量のAPIコールを短時間に発行する必要がなく、スロットリングを構造的に回避できます。

AWS Batchにジョブを投入する方法には、大きく分けて以下の2通りがあります。

  • 個別ジョブ（Single Job） — submitJobを1回呼ぶごとに1個のジョブが実行される、最も基本的な投入方法。グループの数だけsubmitJobを呼び出す必要がある
  • Array Job — submitJobを1回呼ぶだけで、指定した件数（2〜10,000）分の子ジョブをまとめて投入できる仕組み。各子ジョブは環境変数AWS_BATCH_JOB_ARRAY_INDEX（0始まりの連番）で自分が何番目の子ジョブかを判別し、同一のジョブ定義・コンテナイメージを使って並列に実行される

今回のように「グループ数が可変」で「グループ数に比例してジョブ投入APIの呼び出し回数が増えてしまう」という課題に対しては、submitJobの呼び出し回数そのものを1回に抑えられるArray Jobが有力な解決策に見えました。

もう一つの検討ポイント: 直列処理の保証もArray Jobに任せられないか

ここまでの話はあくまで「ジョブ投入APIの呼び出し回数を減らしたい」という課題です。一方で、元々の構成にはもう一つ別の要件がありました。グループ内の項目処理順序を保証する、という要件です（Inline MapのMaxConcurrency: 1で担保していたものです）。

Array Jobを導入するなら、「1グループ＝1個別ジョブ」をループ投入する構成から脱却できるだけでなく、グループ内の処理そのものもArray Jobの子ジョブとして分割（並列化）できるのではないか、という期待もありました。これは先ほどのAPIスロットリングの課題とは別の、順序保証に関する検討ポイントです。

そこで、以下の2点を自分たちで調査することにしました。

気になった2点を調査

  1. AWS Batchは内部キューによる段階的スケジューリングで、APIスロットリングを自動的に回避できるか（＝1つ目の課題の解決策として機能するか）
  2. グループ内の処理を、Array Jobを使って複数コンテナに分割（並列化）できないか（＝2つ目の課題の解決策になり得るか）

調査の結果わかったことは以下の通りです。

  • Array Jobは全インデックスが同一ジョブ定義で並列実行される仕組みであり、処理順序を保証する機能ではありません。公式ドキュメントでも、子ジョブは "a collection of related yet separate basic jobs that might be distributed across multiple hosts and might run concurrently"（並列実行されることがある、関連するが別個の基本ジョブの集合）と説明されています。順序を保証したい場合は、SEQUENTIAL型またはN_TO_N型の依存関係を別途指定する必要があります
  • グループごとに項目数が不均一なため、仮にSEQUENTIAL依存で順序を保証したとしても、Array Jobでグループ単位の不均一な処理量を表現するのはそもそも向かないと判断しました
  • 長時間タスクには、中断リスクのあるFargate Spotではなく、Fargate On-Demandを使うべきと判断しました。Fargate Spotで容量が回収される際は、タスク停止の2分前に警告（SIGTERM）が送られる仕様のため、2分以内に処理を打ち切れない長時間タスクには適しません
  • 起動レートについては自分たちで実測し、1,000ジョブ投入で約69秒（約14〜15タスク/秒）という目安を得ました
  • マルチコンテナのジョブ定義に対応しており、複数コンテナで構成される既存のタスク構成もそのまま移行可能なことを確認しました

ここで重要だったのは、「Array Jobを使えば直列処理の要件ごと解決する」という期待は外れたという点です。Array Jobはあくまで並列実行の仕組みであり、グループ内の順序保証は別途自分たちで設計する必要があることが、調査によって明確になりました。

最終的な移行方法

直列処理の保証をAWSのマネージド機能に頼るのではなく、コンテナ自身の実装で担保する方式に設計を変更しました。

  • 前段の処理の出力形式を「1グループ＝1オブジェクト（グループ内の処理対象リスト付き）」に変更
  • 実行コンテナが起動時にS3から自グループ分の処理対象リストを取得し、コンテナ内のループで順番に処理する
  • チェックポイントは新規にS3ファイルを設けず、既存のステータス管理用Amazon DynamoDBテーブルを流用
  • エラーハンドリングもコンテナ内で完結させ、外部の専用エラーハンドリングLambdaへの依存を解消

グループごとの不均一な処理量への対応をコンテナ側に寄せたことで、「1グループ＝1個別ジョブをループ投入する」必要がなくなり、1回のsubmitJob.sync呼び出しでAWS BatchのArray Jobを投入する方式に切り替えられました。ArrayProperties.Sizeにグループ数を指定し、各子ジョブはAWS_BATCH_JOB_ARRAY_INDEX（0始まり）から自分の担当グループを判別します。

ただしArray Jobはサイズ2〜10,000という制約があり「1」を指定できないため、グループ数に応じてステートマシン側で処理を分岐させています。

グループ数 処理内容
0 終了
1 通常のsubmitJob.sync
2以上 Array JobとしてのsubmitJob.sync

既存の複数コンテナ構成（メイン処理用＋補助処理用）も、AWS Batchのジョブ定義（EcsProperties.TaskProperties、dependsOnで起動順を制御）にそのまま移植できました。なお、.sync統合はEcsPropertiesを使うジョブ定義では非対応だったため、describeJobsをポーリングする形のウェイトループを別途実装しています。

移行後の全体の流れをまとめると、以下のようになります。ポイントは、ステートマシン／AWS Batchが制御する範囲と、コンテナ内部でアプリケーションが制御する範囲がはっきり分かれている点です。

flowchart TD
    Start(["前段の処理の出力<br/>1グループ = 1オブジェクト（処理対象リスト付き）"]) --> Branch{"グループ数は？"}
    Branch -->|0件| End1(["終了"])
    Branch -->|1件| Single["submitJob.sync<br/>通常のジョブとして1件投入"]
    Branch -->|2件以上| Array["submitJob.sync<br/>Array Jobとして投入<br/>ArrayProperties.Size = グループ数"]

    Single --> Poll
    Array --> Poll["ステートマシン側で describeJobs をポーリングし、<br/>全子ジョブの完了を待つ"]

    Array -.->|Array Indexごとに起動| ContainerBoundary
    Single -.->|起動| ContainerBoundary

    subgraph ContainerBoundary["ここから先はコンテナ内部（アプリケーション側）の制御"]
        direction TB
        Idx["AWS_BATCH_JOB_ARRAY_INDEX から<br/>自分の担当グループを判別"] --> Fetch["S3から自グループの<br/>処理対象リストを取得"]
        Fetch --> Loop["リストの項目を先頭から順番にループ"]
        Loop --> Check{"DynamoDBで<br/>この項目の状態を確認"}
        Check -->|未処理 or リトライ対象| Process["処理を実行し、結果を<br/>DynamoDBに記録（チェックポイント）"] --> Loop
        Check -->|処理済み| Loop
        Loop -->|全項目完了| Done(["子ジョブ正常終了"])
    end
    style ContainerBoundary stroke-dasharray: 5 5

    Poll --> Result{"失敗した<br/>子ジョブがあるか？"}
    Result -->|なし| Success(["ステートマシン正常終了"])
    Result -->|あり| Handle["専用Lambdaで、DynamoDB上の<br/>未完了項目を「中断」状態に更新"] --> Fail(["ステートマシンをFailで終了"])

図の点線で囲んだ範囲（コンテナ内部）が、今回新しくアプリケーション側の責務として実装した部分です。ステートマシン・AWS Batch側は「ジョブを投入して完了を待つ」「失敗していたら後片付けをする」という大枠の制御だけを担い、グループ内の処理順序・チェックポイント・リトライ判定といった細かい制御はすべてコンテナの中に閉じています。

移行した結果とトレードオフ

良くなった点

  • 処理対象グループ数が増加しても、ECS RunTaskのAPIスロットリングが発生しなくなった
  • グループごとに不均一な処理量があっても、Array Jobで一括投入できるようになった

新たに生じたトレードオフ

  • 直列処理の保証という責務が、AWSのマネージド機能（Step FunctionsのMaxConcurrency）からアプリケーションコードに移った。チェックポイント管理・リトライ上限・エラー種別の判別をすべてコンテナ側で実装する必要が生じた
  • Array Jobは子ジョブが1件でも失敗すると親ジョブ全体が失敗扱いになる仕様のため、ステートマシンの実行結果だけでは「どのグループが失敗したか」が分からなくなった。これに対しては、個別グループ・項目の成否についてはDynamoDBを正とする設計にし、失敗時は専用のLambdaで未完了分を「中断」状態に更新したうえでステートマシンの実行自体を明示的に失敗させる、という設計で補っています

まとめ

今回の移行で得られた教訓は、「マネージドサービスの機能が、自分たちの要件を暗黙に満たしてくれると期待しない」ことです。Array Jobは名前からも並列処理の機能だと分かりますが、「グループ単位の処理を分割できないか」という発想自体は、既存の直列処理要件の延長線上で出てきた自然な疑問でした。実際に公式ドキュメントや実測で裏を取ったことで、その機能では解決しないことが早い段階で明確になり、手戻りなく「順序保証はアプリ側の責務」という設計判断に進めました。

大規模なバッチ処理基盤を設計する際は、クォータの机上確認だけでなく、想定する使い方がサービスの設計思想と合っているかを公式ドキュメントや実測で確認することが、手戻りを減らす近道だと感じています。

遠藤 広也
(エンジニアブログ記事一覧)
(サバワク記事一覧)

アプリケーションサービス本部

2024年中途入社 アプリケーションサービス本部にてAWSを活用したアプリケーション開発に携わっています！
趣味はお酒とバンドです