YAZエンジニアが運営するメディア

はじめに

今回は、AWS Batchを利用して、Amazon S3に保存したカフェの売上CSVを集計する仕組みを構築します。

AWS Batchは、大量のデータ処理や時間のかかる処理などのバッチ処理を、AWS上で実行・管理するためのサービスです。

今回は3日分の売上CSVを用意し、同じジョブ定義・ジョブキュー・コンピューティング環境を利用しながら、処理対象のCSVだけを変更して複数のジョブを実行します。

今回構築する全体像

Amazon S3に保存したCSVを、AWS BatchのJobとして処理します。
Fargate上でECRに保存したDockerイメージを起動し、コンテナ内の
app.py
がCSVを集計します。

集計結果は再びAmazon S3の
output/
へ保存し、実行ログはCloudWatch Logsから確認します。

各サービスの役割

AWS Batchの主な用途

• ・大量のファイルや画像の一括処理（例：商品画像をまとめてリサイズ・圧縮する）
• ・データの変換・加工（例：大量のCSVから必要なデータを抽出・集計する）
• ・長時間かかる処理（例：大量データを使ったシミュレーションを実行する）
• ・定期的な集計処理（例：1日分の売上データを夜間にまとめて集計する）

AWS Batchのメリット

• ・複数の処理をJobとしてまとめて管理できる
• ・同じ処理を複数のデータに対して実行できる
• ・CPUやメモリなど、処理に必要な実行環境を設定できる
• ・Dockerコンテナを利用して独自のプログラムを実行できる
• ・Fargateと組み合わせることで、EC2インスタンスを自分で用意・管理せずに実行できる

構築手順

Step1 売上データを準備する

① S3バケットを作成する

設定内容

1. AWSマネジメントコンソールからAmazon S3を開きます。
2. ［バケットを作成］をクリックします。
3. バケット名を入力します。
4. その他は基本的にデフォルトのまま作成します。

② 売上CSVをアップロードする

作成したS3バケット内にinputフォルダを作成し、3日分の売上CSVをアップロードします。

input/
├─ orders_20260818.csv
├─ orders_20260819.csv
└─ orders_20260820.csv

CSVの形式

3つのCSVは、すべて同じ列構成で作成します。

CSVの例

menu,category,price,quantity
Cafe Latte,Drink,500,2
Cafe Mocha,Drink,550,1
Cheesecake,Food,600,2
Croissant,Food,350,3
Cafe Latte,Drink,500,1

同じ商品が複数行ある場合は、後ほど実行するapp.pyで商品ごとに販売個数と売上金額を合計します。

集計結果の保存先としてoutputフォルダも作成します。

S3バケット
├─ input/
│   ├─ orders_20260818.csv
│   ├─ orders_20260819.csv
│   └─ orders_20260820.csv
└─ output/

Step2 売上集計プログラムを準備する

① app.pyを作成する

S3から売上CSVを取得し、商品ごとの販売個数と売上金額を集計して、結果CSVをS3へ保存するPythonプログラムを作成します。

import boto3
import csv
import os
import sys
from collections import defaultdict

s3 = boto3.client("s3")
BUCKET_NAME = os.environ["BUCKET_NAME"]

if len(sys.argv) < 2:
    raise ValueError("処理対象のCSVファイル名を指定してください。")

input_filename = sys.argv[1]

INPUT_KEY = f"input/{input_filename}"
OUTPUT_KEY = f"output/report_{input_filename.replace('orders_', '')}"

local_input = f"/tmp/{input_filename}"
local_output = f"/tmp/report_{input_filename.replace('orders_', '')}"

print("=== Batch Brew Cafe 売上集計開始 ===")
print(f"対象ファイル: {INPUT_KEY}")

s3.download_file(
    BUCKET_NAME,
    INPUT_KEY,
    local_input
)

sales = defaultdict(lambda: {
    "quantity": 0,
    "sales": 0,
    "category": ""
})

with open(local_input, encoding="utf-8-sig") as f:
    reader = csv.DictReader(f)

    for row in reader:
        menu = row["menu"]
        category = row["category"]
        price = int(row["price"])
        quantity = int(row["quantity"])

        sales[menu]["category"] = category
        sales[menu]["quantity"] += quantity
        sales[menu]["sales"] += price * quantity

ranking = sorted(
    sales.items(),
    key=lambda x: x[1]["sales"],
    reverse=True
)

with open(local_output, "w", newline="", encoding="utf-8-sig") as f:
    writer = csv.writer(f)

    writer.writerow([
        "rank",
        "menu",
        "category",
        "quantity",
        "total_sales"
    ])

    for rank, (menu, data) in enumerate(ranking, 1):
        writer.writerow([
            rank,
            menu,
            data["category"],
            data["quantity"],
            data["sales"]
        ])

s3.upload_file(
    local_output,
    BUCKET_NAME,
    OUTPUT_KEY
)

total_sales = sum(data["sales"] for data in sales.values())
total_quantity = sum(data["quantity"] for data in sales.values())

print(f"商品種類数: {len(sales)}")
print(f"販売個数: {total_quantity}")
print(f"総売上: {total_sales:,}円")

if ranking:
    print(f"売上1位: {ranking[0][0]}")

print(f"出力先: s3://{BUCKET_NAME}/{OUTPUT_KEY}")
print("=== 売上集計完了 ===")

コード内で行っている処理

• ・Job実行時に処理対象のCSVファイル名を受け取る
• ・S3からCSVをダウンロードする
• ・商品ごとの販売個数と売上を集計する
• ・売上順にランキングを作成する
• ・結果CSVをS3へ保存する

② Dockerfileを作成する

FROM public.ecr.aws/docker/library/python:3.12-slim
WORKDIR /app
RUN pip install --no-cache-dir boto3
COPY app.py .
CMD ["python", "app.py"]

Step3 DockerイメージをAmazon ECRへ登録する

① ECRリポジトリを作成する

Amazon ECRを開き、今回使用するDockerイメージを保存するリポジトリを作成します。

② CloudShellへファイルをアップロードする

AWSマネジメントコンソール上部からCloudShellを開きます。

CloudShellの「アクション」→「ファイルのアップロード」を選択し、
app.pyとDockerfileをそれぞれアップロードします。

アップロード後、以下を実行します。

ls

以下の2ファイルが表示されれば準備完了です。

app.py
Dockerfile

③ Dockerイメージを作成しECRへPushする

DockerイメージのBuild、Amazon ECRへのログイン、タグ付け、Pushについては、以下の記事の手順を参考にしてください。

今回のハンズオンでは、ECRリポジトリ名を今回用のものに読み替えて実施します。
Push完了後、Amazon ECRのリポジトリを開き、Dockerイメージが登録されていることを確認します。

Step4 IAMロールを準備する

今回は、Fargateがコンテナを実行するための実行ロールと、
コンテナ内のapp.pyがS3へアクセスするためのジョブロールの2種類を作成します。

① 実行ロールを作成する

IAMを開き、「ロール」→「ロールを作成」を選択します。

② ジョブロールを作成する

同様に、IAMの「ロール」→「ロールを作成」からジョブロールを作成します。

Step5 コンピューティング環境を作成する

AWS Batchを開き、「コンピューティング環境」から「作成」を選択し、以下のように設定します。

Step6 ジョブキューを作成する

AWS Batchの「ジョブキュー」から「作成」を選択し、以下のように設定します。

緊急の集計処理    ：900
通常の売上集計    ：500
過去データの集計  ：100

Step7 ジョブ定義を作成する

ジョブ定義では、ジョブを実行するときに使用するDockerイメージやIAMロール、vCPU・メモリなどを定義します。

AWS Batchの「ジョブ定義」から「作成」を選択し、以下のように設定します。

環境変数を設定する

app.pyでは以下のコードで値を取得しています。

BUCKET_NAME = os.environ["BUCKET_NAME"]

Step8 ジョブを実行する

AWS Batchの「ジョブ」から「新しいジョブを送信」を選択します。

DockerイメージやvCPU・メモリなどはジョブ定義の設定をそのまま利用します。
今回はコンテナの上書きにあるコマンドを変更し、処理対象のCSVファイル名を指定します。

Step9 実行結果を確認する

ジョブのステータスが最終的にSUCCEEDEDになれば実行成功です。

SUBMITTED
    ↓
PENDING
    ↓
RUNNABLE
    ↓
STARTING
    ↓
RUNNING
    ↓
SUCCEEDED

CloudWatch Logsを確認する

=== Batch Brew Cafe 売上集計開始 ===
対象ファイル: input/orders_20260818.csv
商品種類数: XX
販売個数: XX
総売上: XX,XXX円
売上1位: XXXXX
出力先: s3://xxxxx/output/report_20260818.csv
=== 売上集計完了 ===

S3の出力結果を確認する

output/
└─ report_20260818.csv

Step10 残り2日分のJobを実行する

ジョブ定義やジョブキューを作り直す必要はありません。
新しいジョブを送信し、処理するCSVファイル名だけ変更します。

今回AWS Batchを使うメリット

今回のハンズオンでは、3日分の売上CSVに対して同じ集計処理を実行しました。

ポイントは、日付ごとに新しいプログラムや実行環境を作成する必要がないことです。

共通のジョブ定義
      ↓
同じ集計プログラム
      ↓
┌────┬─────┐
↓         ↓         ↓
8/18      8/19      8/20
CSV       CSV       CSV
↓         ↓         ↓
Job①      Job②      Job③

実際のシステムではどう使う？

今回のハンズオンでは、AWS Batchの画面から手動で3つのジョブを実行しました。

実際のシステムでは、毎回人がAWS Batchを開いて実行するのではなく、
スケジュールや他のAWSサービスと組み合わせて処理を開始する構成が考えられます。

例：毎日の売上を夜間に集計する

店舗・ECサイト
      ↓
1日分の売上データ
      ↓
   Amazon S3
      ↓
夜間にジョブを実行
      ↓
   AWS Batch
      ↓
大量の売上データを集計
      ↓
   Amazon S3
      ↓
翌朝、集計結果を確認

このように、リアルタイムで処理する必要はないものの、
大量のデータをまとめて処理したい場合にバッチ処理が向いています。

料金について

AWS Batch自体には追加料金はかかりません。
今回の構成では、Fargate・Amazon S3・Amazon ECR・CloudWatch Logsなど、実際に利用した各サービスの使用量に応じて料金が発生します。

作成したリソースを削除する

ハンズオン終了後、今回作成したリソースが不要であれば削除します。

• ・ジョブキュー
• ・コンピューティング環境
• ・ジョブ定義
• ・S3バケット
• ・不要なCloudWatch Logs
• ・Amazon ECRリポジトリ
• ・BatchEcsTaskExecutionRole
• ・BatchCafeJobRole

まとめ

今回は、AWS Batchを利用してAmazon S3に保存したカフェの売上CSVを集計する仕組みを構築しました。

• ・Amazon S3に処理対象のCSVと集計結果を保存する
• ・Amazon ECRに売上集計用のDockerイメージを保存する
• ・AWS Batchで複数のJobを管理する
• ・AWS Fargateでコンテナを実行する
• ・Amazon CloudWatch Logsで実行結果やエラーを確認する

AWS Batchでは、同じジョブ定義やジョブキューを利用しながら、異なるデータを複数のJobとして処理できます。
また、Fargateと組み合わせることで、EC2インスタンスを自分で管理することなく、コンテナを利用したバッチ処理を実行できます。