종합 프로젝트: 배치 처리

재실행 가능하고 관측 가능한 데이터 배치를 설계합니다.

개념 이해하기

업무 배치는 파일을 한 번 읽고 끝나는 스크립트가 아니라 입력 검증, 부분 실패, 중복 실행, 체크포인트와 운영 지표를 가진 프로그램입니다.

입력 한 행의 오류가 전체 작업을 중단할지 별도 오류 파일로 보낼지 업무 정책을 정합니다.

idempotency key와 upsert를 사용해 같은 입력을 재처리해도 결과가 중복되지 않게 합니다.

처리 수·성공·실패·소요 시간과 실행 id를 구조화 로그와 metric으로 남깁니다.

이번 강좌에서

재실행 가능하고 관측 가능한 데이터 배치를 설계합니다.

입력 한 행의 오류가 전체 작업을 중단할지 별도 오류 파일로 보낼지 업무 정책을 정합니다.

작은 정상 사례뿐 아니라 실패·경계 사례까지 검증합니다.

처음에 자주 헷갈리는 부분

예외를 출력만 하고 성공 종료 코드로 끝내면 스케줄러가 실패를 알 수 없습니다.

직접 바꿔보세요중복 id와 음수 금액을 가진 행을 추가하고 성공·실패 집계를 구현하세요.

from dataclasses import dataclass
from time import perf_counter

@dataclass(frozen=True)
class Row:
    id: int
    amount: int

def process(rows: list[Row]) -> tuple[dict[int, int], list[str], dict[str, object]]:
    started = perf_counter()
    saved: dict[int, int] = {}
    errors: list[str] = []
    checkpoint = 0
    for index, row in enumerate(rows, start=1):
        if row.amount < 0:
            errors.append(f"id={row.id}: 음수 금액")
            checkpoint = index
            continue
        # id를 idempotency key로 사용한 메모리 upsert 예제입니다.
        saved[row.id] = row.amount
        checkpoint = index
    metrics = {
        "run_id": "batch-demo-001",
        "processed": checkpoint,
        "succeeded": checkpoint - len(errors),
        "failed": len(errors),
        "elapsed_ms": round((perf_counter() - started) * 1000, 3),
        "checkpoint": checkpoint,
    }
    return saved, errors, metrics

saved, errors, metrics = process([Row(1, 1000), Row(2, -1), Row(1, 1200)])
print("saved", saved)
print("errors", errors)
print("metrics", metrics)

실행 결과격리된 Python 환경

실행 버튼을 누르면 결과가 여기에 표시됩니다.