ZrobDev/etl-pipelines

ETL 파이프라인

PostgreSQL(기본 저장소)로의 배치 수집과 Python(uv, ruff, pytest) 사용. 로컬 변환: polars 2.x 또는 DuckDB 2.0.

절차

1. 그레인(Grain) 정의: 한 행이 무엇을 의미하는지, 어떤 자연 키(Natural Key)로 식별되는지 명시합니다. 키가 없으면 멱등 로드가 불가능하므로, 먼저 키(또는 콘텐츠 해시)를 정의하세요.
2. 로드 패턴(아래 참조)을 선택합니다. 증분 전략을 선택합니다. 오케스트레이터는 마지막에 선택합니다.
3. 동일한 윈도우를 재실행해도 동일한 최종 상태가 되도록 실행 코드를 작성합니다. 이를 증명하세요: 테스트에서 두 번 실행하고 행 수와 체크섬이 변경되지 않았는지 확인합니다.
4. 첫 프로덕션 실행 전에 아래의 관측 가능성(Observability) 검사를 추가합니다.
5. 마지막 체크리스트를 검토합니다.

멱등 로드 패턴

• 행에 안정적인 자연 키가 있고 변경 가능한 경우: INSERT ... ON CONFLICT (key) DO UPDATE (Postgres) 또는 MERGE를 사용합니다. WHERE target.hash IS DISTINCT FROM excluded.hash를 추가하여 변경되지 않은 행이 불필요하게 갱신되지 않도록 합니다.
• 데이터가 전체 파티션(일, 배치 ID)으로 도착하고 단위로 재작성되는 경우: 해당 파티션을 하나의 트랜잭션에서 삭제 후 삽입(delete-insert)합니다. 삭제와 삽입이 별도로 커밋되면 크래시 시 구멍이 발생합니다.
• 전체 테이블을 재구축하는 경우: 스테이징 테이블에 로드하고, 검증한 후, 스왑(하나의 트랜잭션에서 이름 변경)합니다. 읽는 쪽에서 반쯤 로드된 테이블을 볼 수 없습니다.
• 키가 없는 추가 전용(Append-only) 팩트: 결정론적 batch_id/소스 파일 해시를 추가하고, 이미 로드된 배치 ID는 거부하거나 대체하도록 합니다.
• 절대 INSERT를 무작정 수행한 후 나중에 중복을 제거하지 마세요. 중복은 중복 제거가 실행되기 전에 대시보드에 도달합니다.
• COPY를 사용하여 임시/스테이징 테이블에 로드한 후, 하나의 집합 기반 업서트(Upsert)를 수행합니다. Python에서 행 단위 업서트는 10~100배 느립니다.

증분 로드

• 워터마크(Watermark): 소스가 단조 증가(Monotonic)하고 인덱싱되어 있음이 보장되는 컬럼(updated_at, 시퀀스 ID)을 사용합니다. 상태 테이블에 저장하고, 로드와 동일한 트랜잭션에서 기록하며, 절대 별도의 파일에 저장하지 않습니다.
• 워터마크 읽기: > watermark가 아닌 watermark - lookback (예: 1~24시간)으로 읽습니다. 소스 커밋은 타임스탬프 순서가 아닐 수 있으며, lookback + upsert가 늦게 도착하는 행을 흡수합니다. lookback 크기는 추측이 아닌 측정된 지연 시간을 기준으로 설정합니다.
• 워터마크 컬럼이 신뢰할 수 없는 경우 (업데이트 시 갱신되지 않음, 시간대 혼합, 소프트 삭제 누락 등): 신뢰하지 말고 주기적으로 전체 조정(Full Reconcile)을 수행합니다(파티션별 키 집합 차이 또는 개수 + 체크섬).
• 소스에서의 하드 삭제(Hard Delete) 는 워터마크에 보이지 않습니다. 명시적으로 결정하세요: 키 차이로 조정하거나, 수용하고 문서화합니다.
• CDC (Postgres 논리적 복제, Debezium 스타일 로그 리더): 삭제, 서브분 지연, 또는 소스를 타임스탬프로 쿼리할 수 없을 때만 사용합니다. CDC는 슬롯, 보존 위험(유휴 슬롯이 소스 디스크를 채움), 순서/스냅샷 처리를 추가합니다. 그 외에는 워터마크를 고수하세요.

백필(Backfill)

• 모든 작업은 --start와 --end (반개방 구간 [start, end), UTC)를 받아들이고 이로부터 모든 것을 파생시킵니다. 변환 내부에 now()를 사용하지 마세요.
• 백필은 기존 윈도우에 대해 일반 작업을 실행하는 것입니다. 다른 코드가 필요하다면 일반 작업이 멱등하지 않다는 뜻이므로 수정하세요.
• 파티션별로 청크(Chunk)하고, 청크를 독립적으로 실행하며, 청크별 상태를 기록하여 청크 40에서 실패해도 40부터 재개할 수 있도록 합니다.
• 동일한 업서트/파티션 대체를 통해 동일한 대상에 백필합니다. 이중 집계가 없는지 확인하세요: 윈도우 전후에 count(*)와 count(distinct key)를 확인합니다.
• 소스 API/DB에 대해 스로틀(Throttle)을 적용하세요; 백필은 자체 부하 스파이크입니다.

재시도, 타임아웃, "정확히 한 번(Exactly Once)"

• 네트워크를 통한 정확히 한 번 전달은 존재하지 않습니다. 적어도 한 번(At-Least-Once) 전달 + 멱등 쓰기를 구축하세요; 결과는 사실상 한 번(Effectively-Once)이 됩니다.
• 멱등한 단계만 재시도하고, 지수 백오프(Exponential Backoff)와 지터(Jitter)를 적용하며, 시도 횟수 상한과 모든 네트워크 호출에 하드 타임아웃(연결 및 읽기 타임아웃 별도 설정)을 설정합니다. 429/5xx/타임아웃에 재시도하고, 4xx에는 재시도하지 않습니다.
• Retry-After를 존중합니다. 전역 속도 제한 없는 재시도는 장애를 증폭시킵니다.
• 이메일 전송, 결제, 웹훅 게시 등 멱등하지 않은 단계는 호출 전에 저장된 멱등 키(Idempotency Key)로 보호하세요.
• 최종 실패 시 0이 아닌 코드로 종료하고 알림을 보냅니다; 스케줄을 녹색으로 유지하려고 예외를 삼키지 마세요.

스키마 드리프트(Schema Drift)

• 랜딩 계층(Landing Layer) 은 관대합니다(원시 페이로드를 JSON/텍스트와 로드 메타데이터로 저장). 타입 계층(Typed Layer) 은 엄격하며, 타입/필수 필드 위반 시 큰 소리로 실패합니다.
• 새 컬럼: 추가적이며 알림과 함께 허용됩니다. 삭제, 이름 변경, 타입 변경된 컬럼: 실행을 실패시키고 차이점을 표시하세요; NULL로 강제 변환하지 마세요.
• 명시적 스키마(polars 스키마, pydantic, 또는 DuckDB DESCRIBE)를 경계에서 한 번만 검증하고, 변환 전체에 흩뿌리지 마세요.

스크레이프-투-웨어하우스(Scrape to Warehouse)

• 두 단계. 1단계: 원시 응답을 변경 없이 불변하게 가져와 저장합니다(소스, 가져온 시간, 콘텐츠 해시를 키로 하는 객체 경로). 2단계: 원시 데이터를 테이블로 파싱합니다.
• 파서 버그는 저장된 원시 데이터를 다시 파싱하여 수정하며, 재가져오기, 금지 위험, 기록 손실이 없습니다.
• 모든 파싱된 행에 URL, 가져온 타임스탬프, HTTP 상태, 파서 버전을 기록합니다. 파싱은 원시 바이트 + 파서 버전의 순수 함수입니다.
• 가져오기 실패 및 빈 페이지/차단된 페이지(CAPTCHA, 로그인 벽이 있는 200)는 상태 코드만으로가 아니라 콘텐츠 검사로 감지합니다.

관측 가능성(Observability) (최소)

• 단계별로 로그: 읽은 행 수, 쓴 행 수, 거부된 행 수, 지속 시간, 윈도우 시작/종료, 실행 ID. 구조화된(JSON) 라인.
• 불변 조건(Invariant) 단언: 허용 오차 내에서 출력 행 vs 입력 행; 일반적으로 비어 있지 않은 윈도우에서 0행은 실행을 실패시킵니다. 조용한 0행 성공은 가장 흔한 미감지 실패입니다.
• 하트비트(Heartbeat): 성공 시와 시작 시 외부 헬스체크(Healthcheck)에 핑을 보내고, 핑 누락 시 알림(크론 미실행, 호스트 다운 감지). 실패 시에도 실패 엔드포인트에 핑을 보냅니다.
• 최신성(Freshness) 에 대한 알림(로드된 최대 타임스탬프가 임계값보다 오래된 경우), 작업 오류만 알리지 않습니다.
• 실행 로그 테이블 유지(실행 ID, 윈도우, 상태, 개수); 이는 백필 원장(Ledger)으로도 사용됩니다.

오케스트레이터 선택

• Cron (또는 systemd timer) + 단일 스크립트: 한두 개의 독립적인 작업, 작업 간 의존성 없음, 로그를 읽을 수 있는 팀. 기본값. 위의 하트비트와 중복 실행 방지 잠금(flock)이 필요합니다.
• n8n: 소량, SaaS API 간 연결, 웹훅 트리거, 비개발자가 플로우 편집. 대용량 배치, 무거운 변환, 코드 리뷰에는 부적합; 무거운 로직은 Python 스크립트로 작성하고 n8n은 이를 호출하세요.
• Airflow 3.x: 많은 DAG, 작업 간 의존성, 일급 백필, 재시도 및 SLA, 감사 UI. 실제 운영 비용(스케줄러, 메타데이터 DB, 워커)이 발생하므로 소수 작업에는 적합하지 않습니다. Airflow 2에서 3으로의 마이그레이션은 호환성이 깨지는 변경 사항이 있습니다; 스니펫을 복사하기 전에 버전을 확인하세요. dbt on Airflow의 경우 Astronomer Cosmos를 사용하세요.
• dlt: 스키마 추론, 증분 상태, 병합/대체 처리를 자동화하려는 API 또는 데이터베이스 수집에 사용합니다. 페이지네이션 추출 + 워터마크 + 업서트 코드를 직접 작성하는 대신 사용하세요; 그 구성에도 위 규칙을 적용하세요.
• 컨테이너: 시크릿은 /run/secrets/<name>에서 읽는 Compose secrets로 전달하며, 절대 env_var나 env_file을 사용하지 마세요; 공개 포트는 127.0.0.1에 바인드하거나 내부 네트워크에 유지하세요.

벤더 스킬

작업에 dbt 또는 Airflow 관련 내용(모델, 테스트, DAG 작성, 오퍼레이터, Cosmos 설정)이 필요할 때는 즉흥적으로 하지 마세요: 이 저장소의 de-vendors 스킬을 통해 가져온 공식 벤더 스킬을 사용하고, references/landscape-2026-11.md에 있는 버전은 현재 상황을 확인하는 용도로만 사용하세요.

리뷰 체크리스트

• 키와 그레인이 명시되었고, 동일한 윈도우 재실행 시 동일한 상태가 됨 (테스트 완료).
• 삭제와 삽입이 하나의 트랜잭션에서 이루어짐; 워터마크가 로드와 함께 커밋됨.
• Lookback이 설정됨; 늦은 데이터와 삭제에 대한 명시적인 답변이 있음.
• 타임아웃, 백오프가 있는 제한된 재시도, 멱등하지 않은 단계 보호됨.
• 원시 랜딩이 불변; 파싱이 분리 가능하고 재실행 가능함.
• 단계별 행 수가 로깅됨; 0행 및 최신성 알림 존재; 하트비트가 설정됨.
• 스키마 드리프트가 타입 계층에서 큰 소리로 실패함.
• 오케스트레이터 선택이 습관이 아닌 작업 수와 의존성에 의해 정당화됨.

구체적인 SQL 및 Python 스켈레톤: references/patterns.md. 실제 사용 경험에서 얻은 교훈: references/gotchas.md.