대량 작업을 작은 청크로 나누는 방법

목차

• #한 번에 처리할 때 생기는 문제
• #청크가 제한하는 네 가지 범위
• #Offset 대신 안정적인 커서로 읽는다
• #기본적인 순차 청크 처리 구조
• #청크마다 트랜잭션을 끊는다
• #청크 크기는 행 개수가 아니라 비용으로 정한다
• #제한된 병렬 처리와 순서 보장
• #여러 Worker가 같은 테이블을 나눠 처리할 때
• #스트림과 backpressure를 활용하는 경우
• #외부 API의 rate limit에 맞춘다
• #종료와 실패 복구를 청크 경계에 맞춘다
• #측정과 튜닝
• #마무리
• #참고 자료
• #관련 노트

한 번에 처리할 때 생기는 문제

관리자 요청으로 50만 명의 알림 설정을 변경한다고 해보자.

const users = await userRepository.findAllActive();

for (const user of users) {
  await notificationRepository.enableDigest(user.id);
}

코드는 이해하기 쉽지만 데이터 전체를 메모리에 올린다. 행 하나가 객체와 문자열을 포함해 평균 2KB만 되어도 50만 건은 약 1GB다. ORM이 메타데이터와 관계 객체를 붙이면 더 커질 수 있다.

하나의 트랜잭션으로 묶으면 다른 문제가 생긴다.

await database.transaction(async (tx) => {
  const users = await tx.user.findAllActive();

  for (const user of users) {
    await tx.notification.enableDigest(user.id);
  }
});

• 트랜잭션이 수십 분 동안 열린다.
• 갱신한 행의 잠금이 커밋까지 유지된다.
• 실패하면 모든 작업이 롤백된다.
• DB 연결 하나를 오래 점유한다.
• 재시도 범위가 50만 건 전체다.
• MVCC 데이터베이스에서는 오래된 버전 정리에 부담을 줄 수 있다.

반대로 모든 행을 동시에 처리하면 메모리 문제를 네트워크·연결 풀 문제로 바꾼다.

await Promise.all(
  users.map((user) =>
    notificationRepository.enableDigest(user.id),
  ),
);

50만 개 Promise가 만들어지고, DB 연결 풀 앞에 대기열이 쌓이며, 외부 API라면 rate limit을 초과한다.

청크 처리는 한 번에 메모리와 실행 시스템에 들어오는 작업량을 제한한다.

500건 읽기 → 처리 → 커밋 → 메모리 해제
다음 500건 읽기 → 처리 → 커밋 → 메모리 해제

청크가 제한하는 네 가지 범위

청크 크기를 단순 페이지 크기로만 보면 중요한 목적을 놓친다.

메모리 범위

한 번에 메모리에 유지할 입력과 중간 결과 수를 제한한다.

대략적 입력 메모리
= 평균 행 크기 × 청크 크기 × 동시 청크 수

트랜잭션 범위

한 번에 커밋하거나 롤백할 행 수와 잠금 유지 시간을 제한한다.

실패 범위

청크 커밋 전에 실패했을 때 다시 처리할 최대 작업량을 제한한다.

부하 범위

DB 연결, 외부 API 호출, CPU 작업의 동시성을 제한한다.

flowchart LR
    A[Source] -->|최대 500행| B[Input Chunk]
    B -->|동시성 8| C[Processor]
    C --> D[Transaction]
    D -->|commit 후| E[Checkpoint]
    E --> A

청크 크기와 내부 동시성, 동시에 실행하는 청크 수를 별도 값으로 봐야 한다.

type BatchTuning = Readonly<{
  chunkSize: number;
  itemConcurrency: number;
  chunkConcurrency: number;
}>;

chunkSize=500, itemConcurrency=10, chunkConcurrency=1이면 메모리에는 최대 한 청크가 있고 그 안에서 10개만 동시에 처리한다.

Offset 대신 안정적인 커서로 읽는다

가장 직관적인 페이지 처리는 Offset이다.

SELECT id, email, digest_enabled
FROM users
WHERE status = 'active'
ORDER BY id
LIMIT 500
OFFSET 100000;

Offset이 커질수록 DB는 앞 행을 찾고 건너뛰는 비용을 지불할 수 있다. 처리 중 행이 추가·삭제되면 페이지 경계도 이동한다.

1페이지: id 1 ~ 500
처리 후 앞쪽 행 100개 삭제
2페이지 OFFSET 500
실제로는 기존 id 601부터 시작
id 501 ~ 600 누락

마지막 정렬 키 이후를 조회하는 커서 방식이 대량 순회에 안정적이다.

SELECT id, email, digest_enabled
FROM users
WHERE status = 'active'
  AND id > $1
  AND id <= $2
ORDER BY id ASC
LIMIT $3;

• $1: 마지막 처리 ID
• $2: 실행 시작 시 고정한 최대 ID
• $3: 청크 크기

첫 실행은 최소 커서에서 시작한다.

let cursor = 0n;
const boundary =
  await userRepository.findMaxActiveUserId();

정렬 키가 고유하지 않다면 tie-breaker를 추가한다.

SELECT id, created_at, payload
FROM events
WHERE (created_at, id) > ($1, $2)
  AND (created_at, id) <= ($3, $4)
ORDER BY created_at ASC, id ASC
LIMIT $5;

청크 처리와 체크포인트의 자세한 일관성은 배치 작업을 재실행 가능하게 만드는 체크포인트에서 다뤘다.

기본적인 순차 청크 처리 구조

가장 안전한 출발점은 한 청크씩 순차 처리하는 것이다.

type UserCursor = Readonly<{
  id: bigint;
}>;

async function runDigestMigration(
  options: {
    chunkSize: number;
    signal: AbortSignal;
  },
): Promise<void> {
  const boundary =
    await userRepository.captureBoundary();

  let cursor: UserCursor = { id: 0n };

  while (!options.signal.aborted) {
    const users = await userRepository.loadNext({
      after: cursor,
      through: boundary,
      limit: options.chunkSize,
    });

    if (users.length === 0) {
      return;
    }

    await processChunk(users);

    cursor = {
      id: users.at(-1)!.id,
    };
  }
}

한 청크가 완전히 끝난 뒤 다음 쿼리를 실행하므로 자연스러운 backpressure가 있다. 처리 속도가 느려지면 읽기 속도도 같이 느려진다.

빈 청크를 종료 조건으로 사용한다

다음 코드는 마지막 행 ID가 0 같은 falsy 값일 때 문제가 될 수 있다.

do {
  const rows = await loadNextBatch(cursor);
  cursor = rows.at(-1)?.id;
} while (cursor);

행 개수로 종료를 판단하는 편이 명확하다.

while (true) {
  const rows = await loadNextBatch(cursor);

  if (rows.length === 0) {
    break;
  }

  await processChunk(rows);
  cursor = rows.at(-1)!.id;
}

필요한 컬럼만 읽는다

-- 피하고 싶은 형태
SELECT *
FROM users
WHERE id > $1
ORDER BY id
LIMIT $2;

SELECT id, digest_enabled
FROM users
WHERE id > $1
  AND id <= $2
ORDER BY id
LIMIT $3;

청크 크기가 같아도 큰 JSON, 이미지 메타데이터, 사용하지 않는 텍스트 컬럼을 제외하면 네트워크와 메모리를 줄일 수 있다.

청크마다 트랜잭션을 끊는다

청크로 읽어도 바깥에 하나의 트랜잭션을 열면 긴 트랜잭션 문제는 그대로다.

// 청크를 나눴지만 전체가 한 트랜잭션
await database.transaction(async (tx) => {
  while (true) {
    const rows = await loadNext(tx, cursor);
    if (rows.length === 0) break;

    await processChunk(tx, rows);
    cursor = rows.at(-1)!.id;
  }
});

트랜잭션을 청크 안으로 옮긴다.

while (true) {
  const rows = await loadNext(cursor);

  if (rows.length === 0) {
    break;
  }

  const last = rows.at(-1)!;

  await database.transaction(async (tx) => {
    await updateUsers(tx, rows);
    await updateCheckpoint(tx, {
      cursorId: last.id,
      processedCount: rows.length,
    });
  });

  cursor = { id: last.id };
}

처리 결과와 체크포인트를 같은 청크 트랜잭션에서 커밋한다. 한 청크가 실패하면 그 청크만 롤백되고 이전 청크는 유지된다.

행마다 UPDATE하지 않고 집합 연산을 먼저 본다

같은 값으로 조건부 갱신할 수 있다면 애플리케이션 반복보다 SQL 한 문장이 효율적일 수 있다.

UPDATE users
SET
  digest_enabled = true,
  updated_at = now()
WHERE id > $1
  AND id <= $2
  AND status = 'active'
  AND digest_enabled = false;

청크 경계를 ID 범위로 나누고 집합 갱신한다. 다만 ID가 희소하면 “500개 ID 범위”가 “500행”과 같지 않다. 먼저 대상 ID를 조회한 뒤 배열로 갱신할 수도 있다.

UPDATE users
SET
  digest_enabled = true,
  updated_at = now()
WHERE id = ANY($1)
  AND digest_enabled = false;

실제 실행 계획, 파라미터 배열 크기, 잠금 시간을 측정해 선택한다.

청크 크기는 행 개수가 아니라 비용으로 정한다

“항상 500개”는 출발값일 뿐 정답이 아니다. 행마다 크기와 처리 시간이 다르다.

작은 설정 행 500개 ≠ 1KB JSON 행 500개 ≠ 10MB 문서 행 500개

청크 크기를 정할 때 목표를 둔다.

- 청크 처리 시간: 2 ~ 10초
- 메모리 증가: 컨테이너 한도의 안전 범위
- 트랜잭션 시간: 잠금 대기 SLO 이하
- 실패 재처리: 허용 비용 이하
- 외부 API 호출: rate limit 이하

메모리 기반 예산

컨테이너 메모리 1GB에서 애플리케이션 기본 RSS가 400MB이고 안전 여유를 300MB 남기고 싶다면 청크와 중간 결과에 약 300MB만 사용할 수 있다.

available_for_batch
= memory_limit
 - baseline_rss
 - safety_margin

행당 실제 힙 증가를 샘플링해 추정한다.

const before = process.memoryUsage().heapUsed;
const rows = await loadSample(1000);
const after = process.memoryUsage().heapUsed;

const estimatedBytesPerRow =
  Math.max(0, after - before) / rows.length;

GC 시점 때문에 한 번의 측정은 부정확하다. 여러 청크의 heapUsed, RSS, GC pause를 함께 관찰한다.

시간 기반 적응

function nextChunkSize(input: {
  current: number;
  durationMs: number;
  targetMs: number;
  min: number;
  max: number;
}): number {
  const ratio =
    input.targetMs / Math.max(input.durationMs, 1);

  const dampenedRatio =
    Math.min(1.5, Math.max(0.5, ratio));

  return Math.round(
    Math.min(
      input.max,
      Math.max(
        input.min,
        input.current * dampenedRatio,
      ),
    ),
  );
}

급격히 두 배·절반으로 바꾸지 않고 변화 폭을 제한한다. 외부 API 429, DB lock wait, 메모리 압박이 발생하면 별도 신호로 더 빠르게 낮출 수 있다.

제한된 병렬 처리와 순서 보장

순차 처리가 느리면 청크 내부에서 제한된 동시성을 사용할 수 있다.

async function mapWithConcurrency<T, R>(
  items: readonly T[],
  concurrency: number,
  mapper: (item: T) => Promise<R>,
): Promise<R[]> {
  const results = new Array<R>(items.length);
  let nextIndex = 0;

  async function worker(): Promise<void> {
    while (true) {
      const index = nextIndex;
      nextIndex += 1;

      if (index >= items.length) {
        return;
      }

      results[index] = await mapper(items[index]);
    }
  }

  await Promise.all(
    Array.from(
      {
        length: Math.min(
          concurrency,
          items.length,
        ),
      },
      () => worker(),
    ),
  );

  return results;
}

await mapWithConcurrency(
  rows,
  8,
  (row) => transformAndStore(row),
);

이 구현은 예시를 위한 것이다. 운영에서는 오류 취소, 재시도, 공정성, telemetry를 제공하는 검증된 동시성 제한 라이브러리도 고려한다.

순서가 중요한 작업

같은 계정의 이벤트는 순서대로 적용해야 할 수 있다.

event 1: +100
event 2: -30
event 3: account closed

전체를 무작위 병렬 처리하면 순서가 뒤바뀐다. 파티션 키별로는 순차, 파티션 간에는 병렬로 처리한다.

const groups = groupBy(
  rows,
  (row) => row.accountId,
);

await mapWithConcurrency(
  [...groups.values()],
  8,
  async (accountEvents) => {
    const ordered =
      accountEvents.toSorted(bySequence);

    for (const event of ordered) {
      await applyEvent(event);
    }
  },
);

이렇게 하면 서로 다른 계정은 최대 8개까지 병렬로 처리하지만 같은 계정 안의 이벤트는 순서대로 적용한다.

예시를 보여주기 위해 잘못된 형태와 올바른 형태를 함께 두었지만 발행 글에서는 올바른 코드만 남기는 편이 읽기 좋다.

병렬 청크와 체크포인트

청크 A, B, C를 동시에 실행했는데 B가 실패하고 C가 먼저 성공할 수 있다.

A 성공
B 실패
C 성공

체크포인트를 C 끝으로 이동하면 B 범위를 건너뛴다. 해결 방법은 다음 중 하나다.

• 청크는 순차 처리하고 내부 항목만 병렬화한다.
• 완료된 청크를 별도 장부에 기록하고 연속 성공 구간까지만 커서를 이동한다.
• 파티션별 독립 커서를 둔다.
• 작업 큐에서 각 항목을 독립 멱등 작업으로 관리한다.

단일 선형 체크포인트가 있다면 청크 순차 + 내부 제한 병렬성이 가장 단순하다.

여러 Worker가 같은 테이블을 나눠 처리할 때

처리 대상 행에 상태가 있다면 FOR UPDATE SKIP LOCKED로 Worker가 서로 다른 행을 선점할 수 있다.

WITH claimed AS (
  SELECT id
  FROM migration_tasks
  WHERE status = 'pending'
  ORDER BY id
  FOR UPDATE SKIP LOCKED
  LIMIT $1
)
UPDATE migration_tasks AS tasks
SET
  status = 'processing',
  owner_id = $2,
  claimed_at = now()
FROM claimed
WHERE tasks.id = claimed.id
RETURNING tasks.*;

여러 Worker가 동시에 실행해도 이미 잠긴 행은 건너뛰고 다른 행을 가져간다. PostgreSQL 공식 문서도 SKIP LOCKED가 일반적인 일관된 조회에는 부적합하지만 queue-like table의 잠금 경합을 피하는 용도로 사용할 수 있다고 설명한다.

flowchart LR
    Q[(Task Table)]
    Q -->|claim 100| W1[Worker A]
    Q -->|claim 100| W2[Worker B]
    Q -->|claim 100| W3[Worker C]

선점 트랜잭션은 짧게

외부 API 호출 동안 DB 행 잠금을 유지하지 않는다. 짧은 트랜잭션에서 상태와 소유자를 기록하고 커밋한 뒤 처리한다.

BEGIN
  pending 행 선점
  status=processing, owner_id 저장
COMMIT

외부 처리

BEGIN
  status=succeeded 저장
COMMIT

Worker가 중간에 죽으면 processing 행이 남는다. claimed_at과 heartbeat로 stale 작업을 찾아 다시 pending으로 돌리거나 새 attempt를 만든다.

소유권 확인

늦게 돌아온 이전 Worker가 새 Worker의 결과를 덮지 않게 상태 갱신 조건에 owner나 attempt를 포함한다.

UPDATE migration_tasks
SET
  status = 'succeeded',
  finished_at = now()
WHERE id = $1
  AND status = 'processing'
  AND owner_id = $2
  AND attempt = $3;

영향받은 행이 0개라면 소유권을 잃은 것이므로 성공 처리하지 않는다.

스트림과 backpressure를 활용하는 경우

파일 변환이나 DB cursor가 스트림을 제공한다면 전체 청크 배열을 직접 관리하지 않고 pipeline을 구성할 수 있다.

import { pipeline } from "node:stream/promises";
import { Transform, Writable } from "node:stream";

const normalizeTransform = new Transform({
  objectMode: true,
  transform(record, _encoding, callback) {
    try {
      callback(null, normalizeRecord(record));
    } catch (error) {
      callback(error as Error);
    }
  },
});

const destination = new Writable({
  objectMode: true,
  highWaterMark: 100,
  async write(document, _encoding, callback) {
    try {
      await searchIndex.upsert(document);
      callback();
    } catch (error) {
      callback(error as Error);
    }
  },
});

await pipeline(
  sourceRecordStream,
  normalizeTransform,
  destination,
);

Node.js Stream은 소비자가 느릴 때 생산자가 계속 버퍼를 늘리지 않도록 backpressure를 제공한다. highWaterMark는 엄격한 메모리 상한이 아니라 읽기를 멈추기 시작하는 임계값이라는 점을 주의한다.

write()의 반환값을 무시하지 않는다

for await (const row of source) {
  destination.write(serialize(row));
}

write()가 false를 반환했는데 계속 쓰면 내부 버퍼가 늘어난다.

import { once } from "node:events";

for await (const row of source) {
  const canContinue =
    destination.write(serialize(row));

  if (!canContinue) {
    await once(destination, "drain");
  }
}

destination.end();

가능하면 pipeline()을 사용해 backpressure와 오류 정리를 맡긴다.

스트림도 체크포인트 문제를 없애지 않는다

DB cursor가 10만 번째 행을 읽었다고 외부 목적지에 10만 번째 결과까지 확정된 것은 아니다. 목적지 쓰기 성공과 내구성 있는 위치 저장을 연결해야 한다. 재시작이 중요한 작업이라면 스트림 내부에서도 일정 개수마다 결과를 flush하고 체크포인트를 커밋한다.

외부 API의 rate limit에 맞춘다

외부 API가 초당 100회를 허용한다고 동시성 100이면 충분한 것은 아니다. 요청이 100ms에 끝나면 초당 약 1,000회가 될 수 있다.

대략적 처리량
≈ concurrency / 평균 요청 시간(초)

평균 200ms, 동시성 10이면 초당 약 50회다. 실제 지연 분포와 재시도를 포함해 여유를 둔다.

토큰 버킷 형태의 rate limiter를 적용할 수 있다.

class TokenBucket {
  #tokens: number;
  #lastRefillAt = Date.now();

  constructor(
    private readonly capacity: number,
    private readonly refillPerSecond: number,
  ) {
    this.#tokens = capacity;
  }

  async take(signal: AbortSignal): Promise<void> {
    while (!signal.aborted) {
      this.#refill();

      if (this.#tokens >= 1) {
        this.#tokens -= 1;
        return;
      }

      await abortableDelay(20, signal);
    }

    throw signal.reason;
  }

  #refill(): void {
    const now = Date.now();
    const elapsedSeconds =
      (now - this.#lastRefillAt) / 1000;

    this.#tokens = Math.min(
      this.capacity,
      this.#tokens +
        elapsedSeconds * this.refillPerSecond,
    );

    this.#lastRefillAt = now;
  }
}

실제 운영에서는 검증된 limiter 라이브러리와 분산 환경의 전역 제한 여부를 검토한다. Worker마다 초당 100회로 설정하고 Worker가 10개면 전체는 초당 1,000회다.

429 응답의 Retry-After를 존중하고 jitter가 있는 backoff를 사용한다.

const delayMs =
  baseDelayMs * 2 ** attempt +
  Math.floor(Math.random() * jitterMs);

재시도 작업도 동시성 슬롯을 차지하게 해 새 요청과 합쳐 제한을 넘지 않도록 한다.

종료와 실패 복구를 청크 경계에 맞춘다

SIGTERM을 받았을 때 새 청크를 가져오지 않고 현재 청크를 완료한 뒤 종료한다.

const shutdownController = new AbortController();

process.once("SIGTERM", () => {
  shutdownController.abort(
    new Error("shutdown requested"),
  );
});

while (!shutdownController.signal.aborted) {
  const rows = await loadNextChunk(cursor);

  if (rows.length === 0) {
    break;
  }

  await processAndCommitChunk(
    rows,
    shutdownController.signal,
  );
}

AbortSignal을 청크 처리 내부에 전달할 때 원자성 정책을 정한다.

soft stop:
  현재 청크 완료 후 종료

hard stop:
  외부 호출과 쿼리를 취소하고 현재 청크 롤백

종료 grace period가 30초인데 한 청크 p99가 2분이면 soft stop은 강제 종료를 피하지 못한다. 청크 시간을 줄이거나 더 긴 grace period를 설정한다.

실패 후 재개

커서는 메모리 변수가 아니라 DB 체크포인트에서 다시 읽는다.

const checkpoint =
  await checkpointStore.load(logicalRunId);

let cursor = checkpoint.cursor;

처리 결과와 체크포인트가 같은 트랜잭션으로 커밋되었다면 마지막 미커밋 청크부터 재개한다. 외부 부작용은 다시 호출될 수 있으므로 멱등하게 만든다.

청크 오류 정책

한 행의 잘못된 데이터가 전체 작업을 영원히 막을 수 있다.

strict:
  한 행 실패 → 청크 롤백 → 작업 실패

quarantine:
  검증 실패 행을 격리 테이블에 기록 → 나머지 진행

retry:
  일시 오류만 제한 횟수 재시도

데이터 정확성이 중요한 금융 갱신은 strict가 적합할 수 있고, 검색 색인 재구축은 잘못된 한 문서를 quarantine하고 진행할 수 있다. 격리 행을 조용히 버리지 않고 알림과 재처리 경로를 둔다.

INSERT INTO batch_quarantine (
  logical_run_id,
  source_id,
  error_code,
  payload_reference,
  created_at
)
VALUES ($1, $2, $3, $4, now())
ON CONFLICT (
  logical_run_id,
  source_id
)
DO UPDATE SET
  error_code = EXCLUDED.error_code,
  created_at = EXCLUDED.created_at;

원본 payload 전체가 개인정보를 포함한다면 격리 테이블에 복사하지 않고 안전한 참조와 오류 코드만 저장한다.

측정과 튜닝

청크 작업에는 평균만으로 부족하다. 큰 행이나 느린 외부 호출 때문에 꼬리가 길어질 수 있다.

청크별 로그

logger.info(
  {
    logicalRunId,
    chunkSequence,
    chunkSize: rows.length,
    cursorStart,
    cursorEnd,
    fetchDurationMs,
    processDurationMs,
    commitDurationMs,
    memoryRssBytes:
      process.memoryUsage().rss,
  },
  "batch chunk completed",
);

원본 ID 범위가 민감하거나 고카디널리티라면 로그에만 두고 메트릭 라벨로 사용하지 않는다.

메트릭

• batch_chunk_duration_seconds{job_name,phase}
• batch_chunk_size{job_name}
• batch_rows_processed_total{job_name,status}
• batch_items_inflight{job_name}
• batch_retry_total{job_name,reason}
• batch_checkpoint_age_seconds{job_name}
• batch_rate_limit_wait_seconds{job_name}
• batch_lock_wait_seconds{job_name}

튜닝 순서

1. 동시성 1, 작은 청크로 정확성을 검증한다.
2. 읽기·처리·커밋 시간을 분리해 병목을 찾는다.
3. 메모리와 트랜잭션 시간이 안전하면 청크 크기를 늘린다.
4. I/O 대기가 병목이면 내부 동시성을 조금씩 늘린다.
5. DB lock wait, 외부 429, 오류율이 늘면 직전 값으로 낮춘다.
6. 장애 주입과 종료 테스트로 복구 범위를 확인한다.

청크 크기와 동시성을 동시에 크게 바꾸면 어느 변화가 영향을 줬는지 알기 어렵다.

예시 비교

숫자는 가상 예시다. 처리량만 보면 마지막 설정도 빨라 보이지만 오류와 메모리, 종료 시간이 나빠진다. 가장 높은 순간 처리량이 아니라 안정적으로 유지되는 처리량을 선택한다.

운영 체크리스트

• 입력에 완전하고 안정적인 ORDER BY가 있는가?
• 실행 시작 상한을 고정해 끝이 무한히 밀리지 않는가?
• 필요한 컬럼만 읽는가?
• 트랜잭션이 청크 범위로 제한되는가?
• 결과와 체크포인트가 같은 원자적 경계에 있는가?
• Promise.all 대신 제한된 동시성을 사용하는가?
• 병렬 완료 순서가 체크포인트 누락을 만들지 않는가?
• 스트림의 backpressure와 오류를 처리하는가?
• 외부 API 제한이 Worker 전체 기준으로 적용되는가?
• 청크 p99 시간이 종료 grace period보다 짧은가?
• 검증 실패 행의 격리와 재처리 경로가 있는가?
• 처리량뿐 아니라 RSS, lock wait, 오류율을 함께 보는가?

마무리

대량 작업을 청크로 나누는 목적은 반복문 모양을 바꾸는 데 있지 않다. 메모리, 트랜잭션, 동시성, 실패의 최대 범위를 예측 가능한 크기로 제한하는 데 있다.

안정적인 커서와 실행 상한으로 한 청크씩 읽고, 청크가 끝날 때 결과와 체크포인트를 커밋한다. 다음 청크는 이전 청크가 끝난 뒤 가져와 backpressure를 유지하고, 필요한 경우 청크 내부에서만 제한적으로 병렬 처리한다.

청크 크기는 관습적인 500이나 1,000이 아니라 행 크기, 처리 시간, 잠금, 외부 rate limit, 종료 예산을 측정해 정한다. 이렇게 만들면 수십만 건의 작업도 “한 번에 성공해야 하는 거대한 트랜잭션”이 아니라 작은 단위로 진행하고 멈추고 다시 시작할 수 있는 운영 가능한 시스템이 된다.

참고 자료

• PostgreSQL LIMIT and OFFSET
• PostgreSQL SELECT and SKIP LOCKED
• Node.js Stream

관련 노트

• 배치 작업을 재실행 가능하게 만드는 체크포인트
• Cron 작업의 중복 실행 방지
• Offset pagination과 Cursor pagination 비교
• Node.js 서버의 Graceful Shutdown 구현하기