AWS BatchはSaaSの非同期処理に向いているか?

はじめに

今回は、AWS Batchを実際に構築し、SaaSの非同期処理に向いているか検証したいと思います。

AWS Batchの要素

コンピューティング環境

ジョブの実行環境を定義します。実行環境として、Fargate, EC2, EKSが選択できますが、今回はFargateを使って説明します。
Fargateを選択した場合、最大vCPU率や、ネットワーク設定(VPC/セキュリティグループ)を設定します。
AWS Batchでは、SQSのような同時実行数制御はできませんが、コンピューティング環境単位では、ここで設定する最大vCPU数を超えることは無いため、ジョブのvCPU数と調整することで、簡易的には実現できます。

ジョブキュー

ジョブが送信されると、コンピューティング環境で実行されるまでの間、ジョブキューに保持されます。
接続されたコンピューティング環境にリソースの空きがある場合、コンピューティング環境でジョブが実行されます。
優先度やスケジュールポリシーを設定することで、ジョブの実行順やリソースの使用量を割り当てることができます。(今回は説明しません。)
- おそらく、コンピューティングリソースを無駄なく使うための機能だと思われます。

ジョブ定義

(Fargateを使用する場合の説明です。)

ジョブが実行される際のDockerイメージや実行コマンド、実行ロールを指定します。(ECSのタスク定義と同じような設定ができ、実際にECSタスクとして実行されます。)
必要とするvCPU数やメモリを指定します。
ジョブが失敗した時の再試行数や、失敗とみなす条件 (Exit Code) を指定することができます。
- 再試行時に別のコマンドを実行することも可能です。

ジョブ

作成したジョブ定義をもとに作成され、ジョブキューに送信します。
ジョブ定義の内容を上書きできます。例えば、コマンドにパラメータを渡してジョブごとに挙動を変化させる事ができます。
ジョブの依存関係を設定することで、前処理が完了してからジョブを実行させることができます。(依存できるジョブは最大20個)

AWS Batchの構築

AWS Batch ダッシュボードの左メニューに有る「ウィザード」を使用すると、ステップに沿って環境構築を進めることができます。

事前準備

今回はFargateを使用します。そのため、事前に実行されるDockerイメージをECRにPushしておく必要があります。
今回作成したDockerイメージはこちらです。

 FROM golang:1.21-alpine WORKDIR /app COPY *.go ./ # Build RUN go build main.go # Run CMD ["./main"]

FROM golang:1.21-alpine

WORKDIR /app

COPY *.go ./

# Build

RUN go build main.go

# Run

CMD ["./main"]

main.goの中身はこちらです。コマンドオプション -batchType を受け取ることができるようになっています。

 package main import ( "flag" "fmt" "os" "time" ) func main() { batchType := flag.String("batchType", "", "Batch Type") flag.Parse() if batchType == nil || *batchType == "" { fmt.Println("Please provide batchType") os.Exit(1) } fmt.Println("Executing many many long tasks...", *batchType) time.Sleep(10 * time.Second) fmt.Println("Finished !!!") }

package main

import (

"flag"

"fmt"

"os"

"time"

)

func main() {

batchType := flag.String("batchType", "", "Batch Type")

flag.Parse()

if batchType == nil || *batchType == "" {

fmt.Println("Please provide batchType")

os.Exit(1)

}

fmt.Println("Executing many many long tasks...", *batchType)

time.Sleep(10 * time.Second)

fmt.Println("Finished !!!")

}

これらを用意し、ローカル環境でdocker buildした上で、ECRに作成したリポジトリにPushします。

コンピューティング環境の作成

まずは、実行環境を選びます。GPUが必要など、特別なケースでなければ、Fargateが適していると思います。

コンピューティング環境設定では、名前を設定します。
インスタンス設定では、最大vCPUを設定します。Fargateを使用する場合は、ジョブごとにECSタスクが立ち上がるのですが、その各ジョブのvCPU数が、ここで指定する最大vCPU数を上回らないように調整してくれます。
もしも、ジョブがキューに送信され、コンピューティング環境のvCPU数にあまりがなければ、ジョブはRUNNABLE状態のままキューに保持されることになります。
このように、最大vCPUを適切に設定することで、ジョブの同時実行数を制御することができます。

ネットワーク設定では、VPCやセキュリティグループなどを選択します。ECSでFargateを使用する場合と同じように、エンドポイントの設定をしておく必要があります。(ECR、S3(ゲートウェイ型)、CloudWatchへのエンドポイントが必要です。)

ジョブキューの作成

ここでは、名前と優先度を設定します。ジョブキューごとに優先度を設定することで、優先度の高いジョブキューに入ったジョブから優先的に処理されます。
ジョブキューは先入れ先出し(FIFO)モデルですが、スケジュールポリシーを使用することでジョブの実行順序をカスタムすることもできます。
SQSほど高機能ではなく、例えばDLQのような機能や、同時実行数の設定などはできません。
(ただし、簡易的で良ければ、同時実行数に関しては最大vCPU数とジョブのvCPU数の調整で、似たようなことは実現できます。)

ジョブ定義の作成

Fargateを使用した、ジョブ定義を作成していきます。基本的には、ECSタスク定義と似たような内容です。

Fargateの設定です。ランタイムプラットフォームの設定や、実行ロールなどを設定します。実行ロールには、ECSのタスク定義と同じく「ecsTaskExecutionRole」が必要です。作成されていない場合は、こちらを参考に作成してください。

次に、ジョブが失敗した時の設定をしていきます。ジョブの試行は、ジョブが失敗した時の試行回数です。デフォルトでは、Exit Codeが0以外の場合に失敗とされ、再試行されます。
また、再試行戦略の条件を設定すると、終了コード等によって、RetryするかExitするか制御することもできます。

コンテナの設定では、使用するDockerイメージやコマンドを設定します。コマンドには Ref::foobar の形でプレースホルダーを書くことができ、ジョブの実行時にこれらをパラメータで上書きすることができます。
パラメータによる上書きは、ジョブ定義にも設定できますが、今回はジョブ実行時に渡せれば良いため、ここでは設定しません。

環境設定では、ジョブロール、vCPU/メモリ、環境変数/シークレットを設定します。
コンテナ内からAWSのリソースにアクセスする場合は、ジョブロールを設定しておきます。
シークレットは、Secret Managerシークレットか、Systems Manager パラメータストアの値が使用でき、いずれの場合でもARNで指定する必要があります。

ジョブの送信

ここまでで作成したリソースを使って、ジョブを実行してみます。今回は、Goからジョブを送信してみます。

 package main import ( "context" "fmt" "log" "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/service/batch" ) func main() { cfg, err := config.LoadDefaultConfig(context.TODO(), config.WithRegion("us-east-2")) if err != nil { log.Fatalf("unable to load SDK config, %v", err) } client := batch.NewFromConfig(cfg) jobName := "batch-go-job" jobDefinition := "batch-go-job-definition" jobQueue := "batch-go-job-queue" output, err := client.SubmitJob(context.TODO(), &batch.SubmitJobInput{ JobName: &jobName, JobDefinition: &jobDefinition, JobQueue: &jobQueue, Parameters: map[string]string{ "batchType": "someJobType", }, }) if err != nil { log.Fatalf("unable to submit job, %v", err) } fmt.Println(output.JobId) }

package main

import (

"context"

"fmt"

"log"

"github.com/aws/aws-sdk-go-v2/config"

"github.com/aws/aws-sdk-go-v2/service/batch"

)

func main() {

cfg, err := config.LoadDefaultConfig(context.TODO(), config.WithRegion("us-east-2"))

if err != nil {

log.Fatalf("unable to load SDK config, %v", err)

}

client := batch.NewFromConfig(cfg)

jobName := "batch-go-job"

jobDefinition := "batch-go-job-definition"

jobQueue := "batch-go-job-queue"

output, err := client.SubmitJob(context.TODO(), &batch.SubmitJobInput{

JobName: &jobName,

JobDefinition: &jobDefinition,

JobQueue: &jobQueue,

Parameters: map[string]string{

"batchType": "someJobType",

})

if err != nil {

log.Fatalf("unable to submit job, %v", err)

}

fmt.Println(output.JobId)

}

clientSubmitJob() 関数を使い、ジョブを送信します。第2引数にジョブの実行に必要な値を渡します。

JobName (必須)
JobDefinition(必須)
- ジョブ定義名もしくはARNを指定。(ジョブ定義のリビジョンを指定することも可能。指定しない場合は最新バージョンが使用される。)
JobQueue(必須)
- ジョブキュー名もしくはARNを指定。
Parameters
- コマンドに Ref::foobar の形で指定したプレースホルダーを置き換える値を指定します。

この他にも、ジョブ定義の内容を上書きすることができます。

ジョブの実行結果は、ダッシュボードから確認することができ、CloudWatch Logsの内容を確認することも可能です。

さいごに

AWS Batchは、以下のような特徴があることが分かりました。

コンピューティング環境の制御(特に最大vCPU数による制御)があるため、コンピューティングリソースを効率よく運用することができる。
ジョブの依存関係を設定することで、並列可能なタスクの前処理・後処理を定義することができる。
ジョブが失敗した場合の再試行や、タイムアウトを定義できる。

AWS Batchを構築するだけで以下の点が実現できるので、SaaSの非同期処理においては向いていると思いました。

長時間の処理ができる。(Lambdaでは15分まで)
標準でキューが実装されているため、大量のリクエストが来る場合に、最大vCPU数の範囲でジョブをさばくことができる。(ECSタスクを直接立ち上げる方法の場合、別途タスク数を調整する方法を考える必要がある。)

逆に、上記のようなケースを実現する必要のない場合はオーバースペックなので、LambdaやECSタスクを直接起動する方法などを検討したほうが良いと思いました。
サービス選定にあたっては、以下の記事も参考になると思います。
https://zenn.dev/faycute/articles/fb310e3ccd783f

日	月	火	水	木	金	土
					1	2
3	4	5	6	7	8	9
10	11	12	13	14	15	16
17	18	19	20	21	22	23
24	25	26	27	28	29	30