Load — Warehouse 또는 downstream 파일에 쓰기 · Data Engineering Quest

Idempotency 가 제일 절실해지는 곳

Extract 와 transform 은 마음껏 retry 해도 돼 — 입력은 read-only 고 출력은 메모리 안에 있으니까. Load 는 달라. 다른 사람이 읽는 목적지에 쓰는 단계라서, 버그 있는 load 는 downstream 리포트, 대시보드, ML 학습 데이터를 오염시켜. 이 stage 에서 낸 실수는 본인 머신 밖으로 퍼져 나가.

네 가지 canonical write 패턴

Replace — target 전체를 덮어써. 동작은 제일 안전한데, 감당되는 건 작은 데이터셋까지야.
Append — 새 row 를 덧붙여. 제일 단순하지만 다시 돌리면 row 가 두 배. 별도 dedup 로직이 없다면 파이프라인에선 금지야.
Partition replace — partition 하나 (예: date=2026-04-30) 를 한 덩어리로 갈아 끼워. 시간으로 partition 된 warehouse 의 default.
Upsert — primary key 로 merge 해. 새 key 면 insert, 있으면 update. 결과가 제일 깔끔한 대신 목적지가 지원해야 해 (Postgres 의 ON CONFLICT, warehouse 의 MERGE).

"Atomic" 은 아껴 써야 하는 단어야

Partition replace 를 보통 atomic swap 이라고 부르는데, 맨 파일시스템 위에서는 아냐. POSIX 가 여기서 주는 원자적 연산은 딱 하나, 경로 하나를 rename 하는 거야. 그리고 그건 이미 내용물이 있는 디렉토리한테는 안 통해. 해 보면 OSError 에 ENOTEMPTY 가 붙어서 나 — errno 값은 macOS 랑 BSD 계열에선 66, Linux 에선 39 야. 숫자를 그대로 비교하지 말고 errno.ENOTEMPTY 를 봐야 하는 이유가 이거고. 그러니까 partition 디렉토리 교체는 언제나 두 번의 연산이야. 진짜 물음은 그 둘을 뭐로 잡느냐지.

지우고 rename 하는 순서를 거의 다 처음에 써. 그게 고를 수 있는 것 중 제일 나쁜 순서야. 그 두 줄 사이엔 partition 이 아예 없고, 거기서 프로세스가 죽으면 옛 데이터는 없앴는데 새 데이터는 안 들어온 상태로 끝나. 옆으로 치우고 rename 하는 순서는 구멍의 길이가 똑같아. 대신 그 안에서 죽어도 옛 partition 이 임시 이름으로 남아 있어서 도로 넣을 수 있어. 비용은 같고 실패는 복구 가능해져. 이게 거래의 전부고, 매번 받을 만한 거래야.

Table format 들이 존재하는 이유이기도 해. Iceberg, Delta Lake, Hudi 는 디렉토리를 바꿔 끼우려고 하지 않아. 새 데이터 파일부터 쓰고, 작은 metadata 파일 하나를 쓴 다음 포인터 하나만 옮겨서 commit 해 — 경로 하나 바꾸기는 진짜로 원자적이니까. Partition 이 커져서 그 구멍이 신경 쓰이기 시작하면, 그때 올라갈 곳이 거기야. 더 영리한 rename 이 아니라.

Code

Partition replace — staging 에 쓰고, 복구 가능한 순서로 swap·python

from pathlib import Path
import shutil
import pandas as pd

def load_partition(df: pd.DataFrame, root: Path, partition: str) -> Path:
    target = root / f'date={partition}'
    staging = root / f'.staging_{partition}'
    retired = root / f'.retired_{partition}'

    if staging.exists():
        shutil.rmtree(staging)
    staging.mkdir(parents=True)
    df.to_parquet(staging / 'part-0.parquet', index=False, compression='zstd')

    # rename 두 번, 절대 지우고 rename 아니고. 이 두 줄 사이에 partition 이
    # 없는 건 어느 쪽이든 같아 — 다만 여기선 옛 사본이 .retired_* 에 남아
    # 있어서, 죽어도 데이터 손실이 아니라 복구가 돼.
    if target.exists():
        if retired.exists():
            shutil.rmtree(retired)
        target.rename(retired)
    staging.rename(target)

    if retired.exists():
        shutil.rmtree(retired)
    return target

load_partition(df, Path('warehouse/orders'), '2026-04-30')

Postgres ON CONFLICT 로 upsert·python

import psycopg

def upsert_orders(rows: list[dict]) -> None:
    sql = '''
        INSERT INTO orders (order_id, customer_id, amount_usd, order_date, updated_at)
        VALUES (%(order_id)s, %(customer_id)s, %(amount_usd)s, %(order_date)s, NOW())
        ON CONFLICT (order_id) DO UPDATE
        SET customer_id = EXCLUDED.customer_id,
            amount_usd  = EXCLUDED.amount_usd,
            order_date  = EXCLUDED.order_date,
            updated_at  = NOW()
    '''
    with psycopg.connect('postgresql://...') as conn:
        with conn.cursor() as cur:
            cur.executemany(sql, rows)

Load — Warehouse 또는 downstream 파일에 쓰기

Idempotency 가 제일 절실해지는 곳

네 가지 canonical write 패턴

"Atomic" 은 아껴 써야 하는 단어야

Code

External links

Exercise

Progress

댓글 0