☀️ haejun

Today-·Total-

대용량 데이터 비동기 배치 파이프라인

7 min read

Summary

  • 대량 데이터 처리는 배치(묶음 처리), 세마포어(동시 개수 제한), 턴스타일(속도 제한), 락(경쟁 상태 방지), 재시도·백오프를 조합한다
  • LLM API는 컨텍스트 캐싱으로 입력 토큰 비용을 줄이고, 배치로 요청 횟수 제한을 우회한다
  • 여러 배치가 하나의 결과 파일에 동시에 쓰면 asyncio.Lock으로 경쟁 상태를 막는다

실전 비동기 디자인 패턴

단순한 asyncio.gather만으로는 API 속도 제한(Rate Limit)이나 서버 부하를 감당하기 어렵다. 대용량 처리는 다음 다섯 패턴을 조합한다. 기본 문법은 비동기 프로그래밍을 참고한다.

배치 처리(Batch Processing) 는 데이터를 하나씩 처리하는 대신 일정량을 묶어 한 번에 처리한다. 네트워크 오버헤드를 줄이고 서버의 병렬 처리 능력을 활용한다. 엘리베이터가 한 명씩이 아니라 10명씩 태우는 것과 같다.

세마포어(Semaphore) 는 공유 자원에 동시에 접근할 수 있는 작업의 최대 개수를 제한한다. 수천 개 코루틴을 전부 동시에 실행하면 서버나 메모리가 폭발하므로 동시 실행 수를 물리적으로 제한한다. 주차장 10칸에 비유된다.

턴스타일(Turnstile) 은 세마포어가 ‘공간’을 제어하는 것과 달리 ‘속도(Rate)‘를 제어한다. Lock과 sleep을 조합해 실행 간격을 강제한다. 1초에 60개를 동시에 던지면 서버가 DDoS로 간주해 429 에러를 뱉으므로, 1초에 1개씩 꾸준히 던진다.

락(Lock) 은 한 번에 하나의 코루틴만 자원에 접근하도록 강제해 경쟁 상태(Race Condition)를 막는다. 공원 화장실 하나에 열쇠 하나인 상황이다.

재시도와 백오프(Retry & Backoff) 는 요청 실패 시 즉시 포기하지 않고 대기 시간을 점진적으로 늘려 재시도한다.

  • 지수 백오프: 대기 시간을 2초 → 4초 → 8초로 배수 증가
  • 지터(Jitter): 대기 시간에 무작위성을 추가해, 모든 요청이 동시에 재시도하며 서버에 2차 충격을 주는 Thundering Herd Problem을 방지

LLM API 효율화 사례

컨텍스트 캐싱

LLM 사용 시 가장 큰 비용은 입력 토큰이다. 매번 똑같은 시스템 프롬프트를 보내는 것은 낭비이므로, 무거운 프롬프트를 서버에 미리 저장(Cache)해두고 요청 시 ID만 보낸다. 캐시된 입력 토큰은 일반 입력보다 훨씬 저렴하고(약 70~90% 할인), 처리 속도도 빨라진다.

from google.generativeai import caching
 
cache = caching.CachedContent.create(
    model="models/gemini-3",
    system_instruction=LONG_SYSTEM_PROMPT,
    ttl=datetime.timedelta(minutes=60)     # 60분간 유효
)

배치 처리로 요청 제한 우회

하루 250회 요청 제한이 있을 때 1회에 이미지 1장씩 보내면 250장이 한계지만, 1회에 20장을 묶어 보내면 5,000장을 처리할 수 있다. 단, 배치 사이즈를 무작정 늘리면 모델의 최대 출력 토큰 제한에 걸려 답변이 잘리므로 주의한다.

파이프라인 템플릿

세마포어(동시 제한), sleep(속도 조절), run_in_executor(블로킹 함수 분리)를 모두 적용한 비동기 배치 처리기다.

import asyncio
import sys
# Windows 'Event loop is closed' 에러 방지
if sys.platform == 'win32':
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
 
async def process_batch(semaphore, model, batch_data):
    async with semaphore:            # 1. 동시 실행 수 제한
        await asyncio.sleep(3.0)     # 2. 속도 조절
        loop = asyncio.get_running_loop()
        # 3. Blocking 함수(업로드)를 Non-blocking으로 변환
        uploaded_file = await loop.run_in_executor(None, sync_upload_func, file_path)
        # 4. 모델 호출
        response = await model.generate_content_async([uploaded_file, prompt])
        return response
 
async def main():
    semaphore = asyncio.Semaphore(5)   # 동시에 5개 배치만
    tasks = [process_batch(semaphore, model, b) for b in batches]
    results = await tqdm.gather(*tasks)

안전한 파일 저장 (Lock)

여러 배치가 거의 동시에 끝나 하나의 결과 파일에 쓰려고 달려들면 파일이 깨질 수 있다. 공유 자원을 건드리는 최소 구간(Critical Section)에만 asyncio.Lock을 건다.

import aiofiles
import asyncio
import json
import os
 
file_lock = asyncio.Lock()   # 전역 락
 
async def append_to_json(file_path: str, new_data: list):
    if not new_data:
        return
 
    async with file_lock:   # 한 번에 하나의 작업만 진입
        if not os.path.exists(file_path):
            current_data = []
        else:
            async with aiofiles.open(file_path, "r", encoding="utf-8") as f:
                content = await f.read()
                current_data = json.loads(content) if content else []
 
        current_data.extend(new_data)
 
        async with aiofiles.open(file_path, "w", encoding="utf-8") as f:
            await f.write(json.dumps(current_data, ensure_ascii=False, indent=2))

Lock 주의사항

  • 병목 주의: Lock을 남발하면 동시성이 사라져 순차 처리처럼 느려진다. 최소 구간에만 건다
  • 데드락 주의: asyncio.Lock은 재진입이 불가능하므로, 락 안에서 또 락을 걸지 않도록 설계한다
sequenceDiagram
    participant Batch_A as 배치 A (완료)
    participant Batch_B as 배치 B (완료)
    participant Lock as file_lock
    participant File as JSON 파일

    Batch_A->>Lock: 락 획득 (Acquire)
    activate Lock
    Lock-->>Batch_A: 승인 (진입)
    Batch_B->>Lock: 락 획득 시도
    Note right of Lock: 이미 잠김! 배치 B 대기
    Batch_A->>File: 읽기 → 추가 → 쓰기
    Batch_A->>Lock: 락 반납 (Release)
    deactivate Lock
    Lock-->>Batch_B: 승인 (배치 B 진입)
    activate Lock
    Batch_B->>File: 읽기(A가 쓴 내용 포함) → 추가 → 쓰기
    Batch_B->>Lock: 락 반납
    deactivate Lock

Reference

Graph View