CSVアップロードをBullMQで非同期処理し、進捗も可視化してみました

はじめに

こんにちは！株式会社BTM 仙台ラボの関口です！
今回はCSVファイルのアップロードからデータのDB保存までの処理について、自分なりに検証・実装してみました。
業務でCSVファイルを扱う機会があったのですが、SaaSのようなアプリで複数のユーザーからアップロードが行われる場合、
どのようにするとアプリに負荷をかけずに済むか気になりました。
CSV をアップロードして DB に取り込む、という処理は一見シンプルに見えますが、

▶　複数のユーザーがアップロードする
▶　ファイルサイズが大きくなる
▶　バリデーションが重い
▶　処理状況をわかりやすくしたい

といった要件が重なると、設計をきちんと考える必要が出てきます。

今回は以下を活用した、CSV アップロードからDB保存までを行う検証用のアプリケーションを実装しました。

▶　署名付き URL による直接アップロード
▶　BullMQ を使ったワーカーによる非同期処理
▶　SSE（Server-Sent Events）によるフロントの進捗表示
▶　pub-subでのワーカーからバックエンドへの通知

以下、それぞれについて順に説明していきます。

全体構成の概要

検証環境 Docker Compose

今回の検証ではDocker Composeにより、以下のような構成を作成しました。

services:
  # --- Vite + React ---
  frontend:
    container_name: csv-processor-frontend
    build:
      context: ./frontend
      dockerfile: Dockerfile
    ports:
      - "5173:5173"
    volumes:
      - ./frontend:/app
      - /app/node_modules
    environment:
      - VITE_BACKEND_URL=${BACKEND_URL}
      - CHOKIDAR_USEPOLLING=true

  # --- Express ---
  backend:
    container_name: csv-processor-backend
    build:
      context: ./backend
      dockerfile: Dockerfile
    ports:
      - "8080:8080"
    volumes:
      - ./backend:/app
      - /app/node_modules
    environment:
      - NODE_ENV=development
      - S3_ENDPOINT=${S3_ENDPOINT_INTERNAL}
      - S3_PUBLIC_ENDPOINT=${S3_ENDPOINT_EXTERNAL}
      - S3_BUCKET_NAME=${S3_BUCKET_NAME}
      - AWS_ACCESS_KEY_ID=${S3_ACCESS_KEY}
      - AWS_SECRET_ACCESS_KEY=${S3_SECRET_KEY}
      - REDIS_HOST=redis
      - REDIS_PORT=6379
    depends_on:
      - minio
      - redis

  # --- S3互換ストレージ ---
  minio:
    image: minio/minio
    container_name: minio-s3
    ports:
      - "${MINIO_API_PORT}:9000"
      - "${MINIO_CONSOLE_PORT}:9001"
    environment:
      MINIO_ROOT_USER: ${MINIO_ROOT_USER}
      MINIO_ROOT_PASSWORD: ${MINIO_ROOT_PASSWORD}
    volumes:
      - minio_data:/data
    command: server /data --console-address ":9001"

  # --- バケット作成 ＋ CORS設定の自動化 ---
  createbuckets:
    container_name: csv-processor-createbuckets 
    image: minio/mc
    depends_on:
      - minio
    entrypoint: ["/bin/sh", "-c"]
    command:
      - >
        until (/usr/bin/mc alias set myminio http://minio:9000 ${S3_ACCESS_KEY} ${S3_SECRET_KEY}) 
　　　　do echo 'Waiting for MinIO...' && sleep 1; done;
        /usr/bin/mc mb --ignore-existing myminio/my-local-bucket;
        /usr/bin/mc anonymous set public myminio/my-local-bucket;
        
        echo '{"CORSRules":[{"AllowedHeaders":["*"],"AllowedMethods"
　　　　:["GET","PUT","POST","DELETE","HEAD"],"AllowedOrigins":["http://localhost:5173"],
　　　　"ExposeHeaders":[]}]}' | /usr/bin/mc cors set myminio/my-local-bucket /dev/stdin;
        
        echo "--- MinIO Setup Finished Successfully ---";

  # --- Redis ---
  redis:
    image: redis:7-alpine
    container_name: csv-processor-redis
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

  # --- Node.js Worker ---
  worker:
    container_name: csv-processor-worker
    build:
      context: ./worker
      dockerfile: Dockerfile
    volumes:
      - ./worker:/app
      - /app/node_modules
    environment:
      - NODE_ENV=development
      - REDIS_HOST=redis
      - REDIS_PORT=6379
      - S3_ENDPOINT=${S3_ENDPOINT_INTERNAL}
      - S3_BUCKET_NAME=${S3_BUCKET_NAME}
      - AWS_ACCESS_KEY_ID=${S3_ACCESS_KEY}
      - AWS_SECRET_ACCESS_KEY=${S3_SECRET_KEY}
      - DATABASE_URL=${DATABASE_URL}
    depends_on:
      - db
      - redis

  db:
    image: postgres:16-alpine
    container_name: csv-processor-db
    environment:
      POSTGRES_USER: ${DB_USER}
      POSTGRES_PASSWORD: ${DB_PASSWORD}
      POSTGRES_DB: ${DB_NAME}
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  minio_data:
  postgres_data:

それぞれのサービスは以下のように責務を分担させています。

▶　frontend: クライアントでの操作、表示
▶　backend: apiサーバー
▶　redis: backendとworkerの連絡経路
▶　worker: csvファイルの検証・DB保存

システム構成

シーケンス図

見やすくするため、SSE接続部等一部省略していますが、全体の処理フローは概ね以下の通りです。

1.署名付きURLでの直接アップロード

MinIO について簡単に

MinIO は S3 互換 API を提供するオブジェクトストレージです。

▶　Dockerイメージがあり、ローカル環境でも使いやすい
▶　AWS SDK がそのまま使える
▶　本番では S3 に移行しやすい

といった理由から、開発・検証用途としても扱いやすく、今回ストレージとして採用しました。

https://www.min.io/

なぜ直接アップロードにしたか

CSV ファイルを一度バックエンドで受け取ってからストレージに保存する構成もありますが、

▶　バックエンドのメモリ消費が増える
▶　同時アップロード数にが増えると、バックエンドのリソース消費が急増する
▶　ファイル転送処理が I/O ボトルネックになりやすい

といった懸念が考えられました。

そこで今回は 署名付き URL（Presigned URL） を使い、
ブラウザから MinIO に直接 PUT する方式を採用しました。

バックエンド側（署名付きURL発行）

app.get('/presigned-url', async (req, res) => {
  const command = new PutObjectCommand({
    Bucket: bucketName,
    Key: fileKey,
    ContentType: fileType,
  });

  const url = await getSignedUrl(s3Client, command, { expiresIn: 300 });
  res.json({ uploadUrl: url, fileKey });
});

フロントエンド側（直接アップロード）

await fetch(uploadUrl, {
  method: 'PUT',
  headers: { 'Content-Type': file.type },
  body: file,
});

この構成により、バックエンドは アップロード処理に関与せず、APIとして専念する 形にできました。

2.BullMQ による非同期ワーカー処理

BullMQ について

BullMQ は、時間のかかる処理や非同期処理を安全に実行するための、
Redis をバックエンドにしたジョブキューライブラリです。

▶Node.jsで使用できて、シンプルな API で扱いやすい
▶リトライや失敗検知が組み込みで用意されており、安定したジョブ実行が可能
▶Queue（ジョブ登録）と Worker（ジョブ実行）の責務が明確で、構成を分離しやすい

今回はbackend, workerでNode.jsを使っていて、
Redisを用意するだけで使える手軽さもあってBullMQを採用しました。

https://bullmq.io/

なぜワーカーに分けたか

CSV のバリデーションや DB 取り込みは、件数によっては数十秒〜数分かかり、
HTTP リクエスト内で完結させるのは難しいので、HTTP とは切り離し、ジョブキュー経由で実行 する形にしました。
また、このような重めの処理は、別のリクエストにも悪影響が及んでしまうので、別プロセスとして隔離しました。

ジョブの登録（Backend）

バックエンド側では、処理したい内容をジョブとして Queue に登録します。
add を呼び出すことで、ジョブは Redis に保存され、Worker が空き次第実行されます。

const job = await fileQueue.add('process-file', {
  fileKey,
  originalName: fileName,
});

第1引数はジョブ名、第2引数は Worker 側で参照できる payload です。
この時点では処理は実行されず、あくまで「処理依頼をキューに積む」だけになります。

ワーカー側の処理（Worker）

Worker は通常、API サーバーとは別プロセス（今回は別コンテナ）で起動し、Queue に積まれたジョブを非同期に処理します。

const worker = new Worker('file-processing-queue', async (job) => {
  const { fileKey, originalName } = job.data;

  // fileKey を使って CSV を取得
  // originalName を使って検証・ログ出力
  // DB への保存処理
});

job.data には、ジョブ登録時に渡した payload がそのまま格納されます。
この例では、バックエンドから渡した fileKey や originalName を使って
ファイル取得や検証処理を行いました。

3.SSE による進捗状況の可視化

SSE について

SSE(Server-Sent Events) は HTTP 接続を維持したまま、サーバーからクライアントへイベントを送信する仕組みです。

https://developer.mozilla.org/ja/docs/Web/API/Server-sent_events

今回のように、アップロードの進捗表示の方法としては、

▶　ポーリング
▶　WebSocket
▶　SSE

などがありますが、今回は

▶　サーバーからクライアントへの一方向通信で十分
▶　余計なリクエストは減らしたい
▶　できるだけシンプルな実装にしたい

という理由から SSE（Server-Sent Events） を選びました。

4.Pub/SubによるWorkerからBackendへの進捗通知

CSVの処理はWorkerが担っているので、workerからbackendへ進捗を知らせる必要がありますが、
この処理でもRedisを活用した ioredis というライブラリを使用し、
Workerからの進捗通知には、Redisを用いたPub/Sub方式を採用しました。

Redisを使用したPub/Subは、Redis公式が出している node-redis というライブラリも使用することができます。
しかし、今回キューで使用したBullMQは、内部でioredisを使用しているため、
ライブラリのメンテナンスの観点も含めてioredisを採用しました。

https://github.com/redis/ioredis?tab=readme-ov-file

バックエンドからDBへのポーリングでは、確認のたびにDB負荷が発生し、
さらにポーリング間隔に起因するタイムラグが避けられません。
Pub/Subによるイベント駆動型の通知にすることで、進捗が更新された瞬間にバックエンドへ情報を伝達でき、
ユーザーに対してリアルタイム性の高い進捗フィードバックを提供することができました。

今回出来たもの

せっかくなので、作成したものの動きを載せておきます。
CSVデータはPythonのFakerライブラリを使用し、ダミーの個人情報を作成しています。
カラムには、名、姓、生年月日、性別、メールアドレスを用意しました。
DBの保存前には簡単にバリデーションを行っています。

進捗がわかりやすいように10万件で試してみます。

ちゃんとプログレスバーで進捗が可視化されてそうです！
DBにもちゃんと10万件登録されていますね。

まとめ

今回の構成では、

▶　バックエンドの負荷分散
▶ユーザー体験

のバランスを取りながら、比較的シンプルな実装でまとめることを意識しました。

キューとワーカーを使用してAPIサーバーとは別のプロセスでタスクを消化するという点は、

CSV 取り込みに限らず、さまざまなバッチ処理系アプリケーションにも応用できそうです。

今回はBullMQ、RedisのPub/Subや署名付きURLの発行・アップロードは初めて自分で実装したので、
処理イメージをつかむことが出来てとても勉強になりました。

今回の実装では、ワーカーのリトライ処理やSSEの再接続等、
異常系をきちんと検証することができなかったので、隙間を見つけてより知識を深めたいです。