Extract — API, 파일, DB 에서 읽기 · Data Engineering Quest

첫 stage 가 제일 어렵다

파이프라인 사고의 대부분은 extract stage 에 몰려 있어 — 본인이 통제하지 못하는 시스템과 코드가 닿는 곳이니까. API 는 throttle 을 걸고, 화요일엔 malformed JSON 을 돌려주고, 연락도 없이 응답 schema 를 바꿔. CSV export 는 vendor 업데이트 한 번에 포맷이 변하고, DB 는 ORM 이 진화하면서 column type 이 흘러가. Extract 의 규율은 이거야: source 가 준 byte 를 있는 그대로 충실히 담고, 뭘 받았는지 기록하고, 타입 싸움은 검증이 맡도록 다음 stage 로 넘겨.

두 패턴

Bulk full extract — 매 run 마다 source 전체를 끌어와. 단순하고, idempotent 하고, state 가 없어. Source 가 작거나 거의 안 변하면 이걸로 충분해.
Incremental extract — 마지막 run 이후 바뀐 것만 가져와. Watermark 컬럼 (updated_at >= last_run) 이나 change feed 를 써. 빠른 대신 watermark 를 관리해야 해.

타협 없는 것들

지수 백오프 retry — transient 에러 (HTTP 429/5xx) 는 기다렸다 다시.
페이지네이션 정확히 — 대부분의 API 는 페이지당 1000 row 이하로 줘. 마지막 페이지가 덜 차서 오거나 API 가 끝났다고 말할 때까지 돌아야 해.
Raw 응답 캡처 — parsed 결과만 남기지 말고. API 가 화요일에 쓰레기를 보냈다면, 디버깅할 때 원하는 건 그 쓰레기 원본이야.
User-Agent 설정 — 본인 파이프라인이라고 밝혀 둬. Upstream 운영자가 트래픽 튀었다고 메일 보낼 때 찾을 수 있어야 하니까.

Code

방어적 페이지네이션 + retry API extractor·python

import time
import logging
from typing import Iterator
import httpx

log = logging.getLogger('extract')

def fetch_orders(client: httpx.Client, since: str) -> Iterator[dict]:
    url = 'https://api.example.com/v1/orders'
    params = {'updated_since': since, 'page_size': 1000}
    page = 1
    while True:
        for attempt in range(5):
            try:
                resp = client.get(url, params={**params, 'page': page})
                if resp.status_code in (429, 500, 502, 503, 504):
                    raise httpx.HTTPStatusError('transient', request=resp.request, response=resp)
                resp.raise_for_status()
                break
            except httpx.HTTPStatusError as e:
                wait = 2 ** attempt
                log.warning('transient error %s, retry in %ds', e.response.status_code, wait)
                time.sleep(wait)
        else:
            raise RuntimeError(f'page {page} retry 다 소진')

        body = resp.json()
        for row in body['results']:
            yield row

        if not body.get('next_page'):
            return
        page += 1

with httpx.Client(
    headers={'User-Agent': 'cwk-de-pipeline/1.0 (ops@example.com)'},
    timeout=30.0,
) as client:
    rows = list(fetch_orders(client, since='2026-04-01T00:00:00Z'))

Exercise

공개 REST API 하나 골라 (GitHub, OpenWeather, 무료 주식 ticker). 페이지네이션을 처리하고, transient 에러를 최소 한 번은 retry 하고, parsing 전에 raw 응답을 disk 에 쓰는 extractor 를 만들어. 그다음 그 raw 파일을 읽어 DataFrame 으로 parse 하는 함수를 따로 만들어. 이 분리 덕분에 두 쪽을 각각 따로 테스트할 수 있다는 걸 확인해 봐.

Extract — API, 파일, DB 에서 읽기