공개 데이터셋·Kafka·웨어하우스를 연결하는 데이터 파이프라인 설계

공개 데이터셋으로 시작한 실험 데이터와 Kafka로 유입되는 운영 이벤트는 성격이 다르다. 이 둘을 같은 적재 경로에 무리하게 넣기보다, 스테이징·검증·웨어하우스 적재를 분리한 파이프라인으로 연결해야 분석과 운영 모두에서 재현성을 확보할 수 있다.

서로 다른 데이터 소스가 맡는 역할

Kaggle과 UCI 같은 공개 저장소는 구조화·반구조화 데이터의 스냅샷을 제공한다. 모델 실험, 벤치마킹, 초기 데이터 확보에 적합하지만 파일 다운로드 이후 검증·버저닝과 스냅샷 기준의 시계열 관리가 필요하다.

Kafka는 사용자 이벤트, 로그, 센서 신호처럼 계속 발생하는 데이터를 전달하는 채널이다. 파티션과 오프셋을 중심으로 내결함성을 구성하며, 토픽의 키 선택과 파티션 수 산정, 스키마 레지스트리 표준화가 운영 설계의 핵심이 된다.

데이터 웨어하우스는 분석과 리포팅을 위해 최적화된 MPP 또는 Managed SQL 엔진이다. ACID, 컬럼너 스토리지, 쿼리 옵티마이저를 기반으로 대규모 스캔과 조인을 처리하고, 데이터 마트·BI·피처 스토어의 분석 기반이 된다.

구분	성능(처리량/지연)	확장성	일관성	안정성	운영 편의
공개 데이터셋 (Kaggle/UCI)	파일 I/O 중심, 병렬 다운로드 시 선형 확장	스토리지/네트워크 의존	스냅샷 기준, 갱신 불규칙	외부 호스트 의존, 미러링 권장	단순 절차, 메타데이터 관리 필요
스트리밍 (Kafka)	고처리량, 저지연(10~100ms), 배치·압축 영향	파티션 수 비례 스케일아웃	At-least-once 기본, Exactly-once는 추가 구성 필요(최신 정보 확인 필요)	복제/ISR 기반, 브로커 장애 내성	클러스터 운영 부담, 모니터링 필요
데이터 웨어하우스	대규모 스캔·조인 최적화, MPP 성능	컴퓨트/스토리지 분리형 스케일	ACID, 스냅샷 격리	고가용성·자동복구	SQL 중심, 관리형 서비스 편의 높음

수집부터 분석 제공까지의 경로

배치 데이터는 병렬 다운로드, 체크섬 검증, 증분 동기화(rsync/S3 sync)를 거쳐 스테이징에 둔다. 스트림에서는 Producer 재시도·압축·배치 크기 조정과 Consumer 그룹 스케일아웃을 함께 고려한다.

스테이징 이후에는 스키마 검증, 정합성 검사, 엔코딩, 파티셔닝을 수행한다. 검증된 데이터는 COPY/LOAD로 웨어하우스에 적재한 뒤 데이터 마트, 피처 스토어, BI에 공급한다.

⤢✕

Kafka에서는 Producer idempotence와 트랜잭션을 활성화하고, Consumer는 at-least-once 방식으로 처리한 뒤 싱크 적재 단계에서 멱등 MERGE를 적용한다. Exactly-once 보장은 브로커·클라이언트 버전과 설정에 의존하므로 최신 정보 확인이 필요하다.

웨어하우스는 스테이징 테이블에 배치를 로드한 다음 대상 테이블과 MERGE 트랜잭션을 수행한다. 커밋 단위는 파티션 또는 배치 기준으로 잡고, 실패하면 전체 롤백한다.

오류도 경로별로 분리한다. 스트림의 역직렬화 실패는 Dead Letter Topic에 격리하고 샘플링 검사 뒤 재주입한다. 배치에서 체크섬이나 행수 검증이 실패하면 재시도 큐에 넣으며, 최대 재시도를 넘기면 알림과 차단을 수행한다. 적재 단계의 제약 조건 위반 행은 별도 격리 테이블에 저장하고 재처리 규칙을 자동화한다.

ETL과 ELT 사이의 선택

ETL은 소스 근처에서 정제와 집계를 마친 후 웨어하우스에 적재하는 방식이다. 초기 품질을 확보하기 쉽지만 변환 비용이 늘어난다.

ELT는 원천 데이터를 먼저 적재하고 SQL로 변환한다. 민첩성과 재현성이 장점인 대신 스토리지와 쿼리 비용을 관리해야 한다. 배치 스테이징을 안정화한 뒤 스트림을 병행하고, 이후 ELT와 데이터 마트를 고도화하는 순서로 확장할 수 있다.

Kafka에서 Parquet로 마이크로배치 저장

환경은 Python 3.10+, confluent-kafka>=2.3, pandas>=2.0, pyarrow>=12를 전제로 한다. Kafka 토픽(events)의 JSON 메시지를 받아 로컬 또는 오브젝트 스토리지에 Parquet 마이크로배치로 저장한다.

# kafka_to_parquet.py
# 목적: Kafka → Parquet 마이크로배치 저장(멱등 파일명, 파티셔닝)
import json, os, time, uuid
from datetime import datetime, timezone
from confluent_kafka import Consumer
import pandas as pd

BOOTSTRAP = os.getenv("KAFKA_BOOTSTRAP", "localhost:9092")
TOPIC = os.getenv("TOPIC", "events")
OUT_DIR = os.getenv("OUT_DIR", "data/landing/events")

def consumer():
    c = Consumer({
        "bootstrap.servers": BOOTSTRAP,
        "group.id": "events-batcher",
        "auto.offset.reset": "earliest",
        "enable.auto.commit": False
    })
    c.subscribe([TOPIC])
    buf = []
    last_flush = time.time()
    FLUSH_SEC = 60
    FLUSH_SIZE = 10_000

    os.makedirs(OUT_DIR, exist_ok=True)

    try:
        while True:
            msg = c.poll(1.0)
            now = time.time()
            if msg is None:
                pass
            elif msg.error():
                # DLQ 설계: 실제 운영 시 DLQ 프로듀서로 전송 권장
                print(f"Error: {msg.error()}")
            else:
                try:
                    rec = json.loads(msg.value())
                    buf.append(rec)
                except Exception as e:
                    print(f"Deserialize error: {e}")

            if len(buf) >= FLUSH_SIZE or (now - last_flush) >= FLUSH_SEC:
                if buf:
                    ts = datetime.now(timezone.utc).strftime("%Y-%m-%d")
                    part = uuid.uuid4().hex[:8]
                    out_path = f"{OUT_DIR}/dt={ts}/part-{part}.parquet"
                    os.makedirs(os.path.dirname(out_path), exist_ok=True)
                    df = pd.DataFrame(buf)
                    # 기본 정규화: 중복 컬럼 제거, 타입 캐스팅 등 추가 가능
                    df.to_parquet(out_path, index=False)
                    print(f"Wrote {len(buf)} records to {out_path}")
                    buf.clear()
                    last_flush = now
                c.commit()  # at-least-once
    finally:
        c.close()

if __name__ == "__main__":
    consumer()

스테이징 데이터를 멱등하게 웨어하우스에 적재하기

stage.events에 Parquet 로드가 끝났고, dw.events가 존재하며, event_id와 event_ts를 키로 사용한다고 가정한다. 다음은 Snowflake/BigQuery 호환 형태(개념 동등)의 예시다.

MERGE INTO dw.events t
USING (
  SELECT DISTINCT event_id, event_ts, user_id, action, payload
  FROM stage.events
) s
ON t.event_id = s.event_id AND t.event_ts = s.event_ts
WHEN MATCHED THEN UPDATE SET
  user_id = s.user_id,
  action  = s.action,
  payload = s.payload
WHEN NOT MATCHED THEN INSERT (
  event_id, event_ts, user_id, action, payload
) VALUES (
  s.event_id, s.event_ts, s.user_id, s.action, s.payload
);

파티션 단위로 COPY/LOAD한 뒤 MERGE를 수행하고, 성공한 경우에만 스테이징을 청소한다. 실패한 배치는 보존하고 재시도할 수 있어야 한다. MERGE 성능은 웨어하우스별 쿼리 힌트와 클러스터 키 기능을 활용해 조정할 수 있다.

거버넌스와 운영에서 확인할 지점

스키마 레지스트리와 backward/forward/full 호환성 규칙을 적용하고, 계약 위반은 CI에서 차단한다. 데이터 품질 검사는 개수, 고유도, null 비율, 참조 무결성을 대상으로 자동화하며 실패 시 차단과 알림을 연결한다.

파티셔닝 키는 dt, org_id 등을 기준으로 설계하고, 소형 파일을 합치는 compaction 배치를 운영한다. 원천과 스테이징에는 최소 권한을 적용하며 민감 필드는 칼럼 암호화 또는 마스킹으로 보호한다. 스토리지 수명주기 자동 만료와 웨어하우스 워크로드 격리는 비용 거버넌스의 일부다.

실시간 목표가 초·분 단위로 좁아질수록 운영과 관측 비용은 증가한다. ETL은 품질을 먼저 반영하는 선택이고, ELT는 민첩성과 재현성을 우선하는 선택이다. Exactly-once 역시 구성·버전 제약이 있어 운영 복잡성이 커지며 최신 정보 확인이 필요하다.

모델 개발에서는 Kaggle/UCI로 초기 학습과 특성 엔지니어링을 시작한 뒤 운영 로그를 결합해 현실 데이터 분포를 보정할 수 있다. 제품 분석에서는 Kafka 클릭스트림을 마이크로배치로 적재해 대시보드 지연을 5분 이내로 유지한다. 과거 판매 배치 데이터와 현재 이벤트 스트림을 함께 사용하면 수요 예측용 융합 피처를 만들 수 있고, 예측 정확도 개선과 재고 회전율 상승으로 이어진다.

생산팀과 schema registry 기반의 스키마 계약을 맺고 배포 전 호환성을 검사하면 API 품질을 관리할 수 있다.

배치와 스트림을 융합하면 대시보드 지연은 60→5분 수준으로 감소한다. 스키마 계약과 DQ 자동화는 적재 실패율을 50% 이상 낮추며, ELT와 컴퓨트 격리는 쿼리 비용을 20~~30% 절감한다. 표준 파이프라인 템플릿과 오케스트레이션은 배포 리드타임을 30~~50% 단축한다.