PySpark 증분 처리 — Iceberg/Delta CDC와 Change Data Feed

택배 앱의 배송 현황을 생각해 봅시다. 현재 위치를 보여 주기 위해 전국 모든 택배 기록을 매분 통째로 다시 읽는다면 어떨까요? 현실에선 그렇게 하지 않습니다. "지금 막 상태가 바뀐 건"만 골라 반영하죠. 데이터 파이프라인도 마찬가지입니다. 데이터가 커지면 전체 테이블을 매번 재처리하는 건 사실상 불가능해집니다. 해법은 증분 처리 — "마지막 처리 이후 바뀐 행만" 읽어 하류로 전파하는 것입니다.

전통적으로 증분 처리는 updated_at 워터마크로 어렵게 구현했지만, 현대 Lakehouse 포맷은 변경 자체를 추적하는 기능(Iceberg incremental read, Delta Change Data Feed)을 내장합니다. 이 글은 PySpark 로 변경분을 읽어 다운스트림에 멱등하게 반영하는 패턴을 정리합니다.

이 글에서 배우는 것

• 증분 처리가 무엇이고 왜 필요한지
• Iceberg 스냅샷 기반 증분 읽기 방법
• Delta Change Data Feed(CDF)로 insert/update/delete를 행 단위로 추적하는 방법
• CDF + MERGE로 다운스트림에 변경을 멱등하게 반영하는 패턴
• 메달리온 아키텍처(Bronze→Silver→Gold) 증분 전파와 집계 시 주의할 점

1. 증분 처리란

핵심은 "무엇이 바뀌었는가(insert/update/delete)"를 아는 것입니다. 증분 처리를 구현하는 방식은 크게 두 가지로 나뉩니다.

2. Iceberg — 스냅샷 기반 증분 읽기

Iceberg 는 매 커밋이 스냅샷입니다. 마치 사진 앨범에서 특정 날짜 이후 찍힌 사진만 골라보듯, 두 스냅샷 사이에 새로 추가된 데이터만 골라 읽을 수 있습니다.

# 특정 스냅샷 이후 추가분만 읽기 (append 스냅샷 대상)
incr = (spark.read
    .format("iceberg")
    .option("start-snapshot-id", last_processed_snapshot)
    .option("end-snapshot-id", current_snapshot)
    .load("analytics.events"))
 
# 또는 타임스탬프 기반
incr = (spark.read
    .format("iceberg")
    .option("start-timestamp", last_run_ts_ms)
    .load("analytics.events"))

처리 후, 마지막으로 본 스냅샷 ID 를 저장해 다음 실행의 시작점으로 씁니다. "지난번엔 여기까지 읽었다"는 책갈피를 남기는 셈이죠.

# 처리 후 현재 스냅샷 ID 를 체크포인트 테이블에 기록 (다음 실행 시작점)
current = spark.sql("SELECT snapshot_id FROM analytics.events.snapshots ORDER BY committed_at DESC LIMIT 1")

Iceberg incremental read 는 주로 append(추가) 를 다룹니다. update/delete 까지 행 단위로 추적하려면 Delta CDF 나 별도 설계가 필요합니다.

3. Delta Change Data Feed (CDF)

Delta 의 CDF 는 insert/update/delete 를 행 단위로 기록합니다. 단순히 "무언가 바뀌었다"가 아니라 "누가 어떻게 바뀌었는지"까지 알 수 있어 훨씬 강력합니다.

# 테이블에 CDF 활성화 (생성 시 또는 ALTER)
spark.sql("ALTER TABLE analytics.users SET TBLPROPERTIES (delta.enableChangeDataFeed = true)")

# 버전(또는 타임스탬프) 범위의 변경분 읽기
changes = (spark.read
    .format("delta")
    .option("readChangeFeed", "true")
    .option("startingVersion", last_version)
    .table("analytics.users"))
 
# changes 에는 특별 컬럼이 추가됨:
#   _change_type: insert / update_preimage / update_postimage / delete
#   _commit_version, _commit_timestamp

CDF 를 읽으면 각 행에 _change_type 컬럼이 붙습니다. 변경 종류는 아래와 같습니다.

스트리밍으로도 변경분을 구독할 수 있습니다.

stream = (spark.readStream
    .format("delta")
    .option("readChangeFeed", "true")
    .table("analytics.users"))

4. 변경분을 다운스트림에 반영 (멱등 MERGE)

변경분을 읽었다면, 이제 하류 테이블에 적용할 차례입니다. insert/update 는 upsert 로, delete 는 실제 삭제로 반영하면 소스와 하류를 항상 일치시킬 수 있습니다.

def apply_changes(batch_df, batch_id):
    # update 는 postimage 만 사용, delete 와 함께 처리
    latest = (batch_df
        .filter("_change_type IN ('insert','update_postimage','delete')"))
    latest.createOrReplaceTempView("changes")
 
    batch_df.sparkSession.sql("""
        MERGE INTO marts.users t USING changes s
        ON t.id = s.id
        WHEN MATCHED AND s._change_type = 'delete' THEN DELETE
        WHEN MATCHED THEN UPDATE SET *
        WHEN NOT MATCHED AND s._change_type != 'delete' THEN INSERT *
    """)
 
(stream.writeStream
    .option("checkpointLocation", "s3://bucket/ckpt/users-cdc")
    .foreachBatch(apply_changes)
    .start())

한 문장으로: CDF + MERGE 는 소스의 변경을 하류에 정확히, 그리고 안전하게 전파하는 표준 패턴입니다.

delete 까지 반영되므로 하류 테이블이 소스와 언제나 일관되게 유지됩니다.

5. 메달리온 아키텍처 — 증분 전파

증분 처리의 진가는 여러 레이어(Bronze→Silver→Gold)를 증분으로 연결할 때 드러납니다. 각 레이어가 "바뀐 것만" 다음 레이어로 흘려보내므로 전체 데이터를 매번 재계산할 필요가 없습니다.

# Silver: Bronze 의 변경분만 읽어 정제 후 Silver 에 머지
bronze_changes = read_changes("bronze.events", last_version)
cleaned = transform(bronze_changes)
merge_into("silver.events", cleaned)
 
# Gold: Silver 의 변경분만 읽어 집계 갱신 (증분 집계는 주의 — 아래)

각 레이어가 변경분만 처리하니, 매일 수 TB 를 통째로 재계산하지 않아도 신선한 데이터를 유지할 수 있습니다.

6. 함정 — 증분 집계의 정합성

증분으로 upsert 를 전파하는 건 비교적 쉽습니다. 그런데 증분 집계는 훨씬 까다롭습니다. "어제 합계 + 오늘 변경분"이 항상 맞지는 않거든요. 특히 update/delete 가 이미 집계된 과거 값을 건드릴 때 문제가 생깁니다.

⚠️ insert-only 스트림의 집계는 증분이 쉽지만, update/delete 가 과거를 바꾸면 해당 파티션/그룹만 재집계하는 게 안전합니다. 영향받는 키만 골라 부분 재계산하는 설계를 고려하세요.

7. 증분 vs 전체 재처리 선택

상황에 따라 전략이 다릅니다. 증분이 항상 옳은 건 아닙니다. 아래 표를 참고해 여러분의 파이프라인에 맞는 방식을 고르세요.

데이터가 작으면 전체 재처리가 단순하고 버그가 적습니다. 복잡도를 도입하기 전에 "정말 증분이 필요한 규모인가?"를 먼저 따져 보세요.

8. 정리

증분 처리의 핵심은 "전체를 다시 읽지 말고, 바뀐 것만 읽어 전파하라"입니다. Iceberg 의 스냅샷 증분과 Delta 의 Change Data Feed 가 변경 추적을 테이블 포맷 차원에서 제공하므로, updated_at 으로 고생하던 증분 로직이 선언적으로 단순해집니다. CDF + MERGE 로 insert/update/delete 를 하류에 멱등하게 반영하고, 메달리온 레이어를 증분으로 연결하면 — 수 TB 를 매일 재계산하지 않고도 신선한 데이터를 유지할 수 있습니다. 단, 증분 집계의 정합성만은 신중히 다루세요.

마치며 — 핵심 요약

• 증분 처리는 전체 재처리 대신 "마지막 이후 바뀐 행만" 골라 처리하는 패턴입니다.
• Iceberg incremental read는 스냅샷 간 append 를 효율적으로 읽고, 스냅샷 ID 를 체크포인트로 관리합니다.
• Delta CDF는 insert/update/delete 를 행 단위로 기록해 변경 유형까지 구분할 수 있습니다.
• CDF + MERGE 조합으로 다운스트림에 변경을 멱등하게 전파하면 소스와 하류가 항상 일치합니다.
• 메달리온 아키텍처에서 각 레이어를 증분으로 연결하면 수 TB 의 일일 재계산 없이 데이터를 신선하게 유지할 수 있습니다.
• 단, 증분 집계는 update/delete 가 끼면 복잡해집니다 — 영향받는 파티션만 부분 재계산하는 전략을 처음부터 설계에 반영하세요.

증분 파이프라인이 처음엔 복잡해 보여도, CDF 와 MERGE 가 변경 추적의 무거운 짐을 대신 져 줍니다. 한 단계씩 적용해 보면 금방 익숙해질 거예요.

이 글은 Spark 3.5 + Iceberg/Delta 기준으로 작성되었습니다. 증분 처리·CDC 파이프라인이나 메달리온 아키텍처 설계가 필요하시면 언제든 문의해 주세요.

— Data Dynamics 엔지니어링 팀