DEV Community

박준현
박준현

Posted on

skip에서 partition overwrite로: business_date 재처리를 Iceberg로 다시 표현하기

#dataengineering #python #etl #learning

skip에서 partition overwrite로: business_date 재처리를 Iceberg로 다시 표현하기

이전 글에서는 같은 source_hash가 다시 들어왔을 때 기존 successful run을 재사용하는 idempotency를 다뤘다.

하지만 재처리에는 두 종류가 있다.

1. 같은 입력이 다시 들어온 경우
   -> skip이 맞다.

2. 같은 business_date의 정정 입력이 들어온 경우
   -> skip하면 안 된다.
   -> 같은 날짜의 gold 결과를 중복 없이 교체해야 한다.
Enter fullscreen mode Exit fullscreen mode

manufacturing-data-platform-mini의 B5 slice는 두 번째 문제를 아주 작게 다룬다.

전체 Spark pipeline을 만든 것이 아니다. gold_daily_metrics Iceberg table 하나를 local Spark에서 만들고, business_date partition overwrite와 snapshot evidence만 검증했다.

Scenario

이미 아래 gold row가 있다.

business_date=2026-06-29
plant-a / line-1 / gearbox-a
units_produced=120
defect_count=3
Enter fullscreen mode Exit fullscreen mode

나중에 같은 business_date=2026-06-29에 대한 정정 source가 들어온다.

운영자가 원하는 것은 append가 아니다.

원하지 않는 상태:
  2026-06-29 old row
  2026-06-29 corrected row
  -> 같은 날짜 결과가 중복됨

원하는 상태:
  2026-06-29 corrected row만 남음
  2026-06-30 같은 다른 날짜 partition은 그대로 유지됨
  재처리 전후 snapshot evidence가 남음
Enter fullscreen mode Exit fullscreen mode

그래서 이 slice의 질문은 이렇다.

같은 business_date의 정정 source를 처리할 때,
gold table에서 해당 날짜 partition만 중복 없이 교체하고,
어떤 run이 어떤 Iceberg snapshot을 만들었는지 남길 수 있는가?
Enter fullscreen mode Exit fullscreen mode

Decision Pressure

Slice1의 CSV pipeline은 already-successful source를 안전하게 skip할 수 있다.

dataset_id + business_date + source_hash
Enter fullscreen mode Exit fullscreen mode

이 key가 같으면 같은 입력이다. 다시 계산해도 같은 결과이므로 기존 run을 재사용한다.

하지만 source_hash가 달라졌다면 의미가 다르다.

same business_date
different source_hash
Enter fullscreen mode Exit fullscreen mode

이건 retry가 아니라 correction이다.

CSV run-folder 방식에서는 새 run output을 만들 수는 있지만, "현재 gold table에서 해당 날짜를 원자적으로 교체한다"는 table-level 의미가 약하다.

Iceberg를 붙이는 이유는 여기 있다.

source_hash
  -> 같은 입력인지 판단하는 idempotency key

business_date partition
  -> 정정 시 교체할 gold table 범위

snapshot_id
  -> table commit의 evidence
Enter fullscreen mode Exit fullscreen mode

즉 Spark/Iceberg는 도구 이름을 추가하려고 붙인 것이 아니라, 재처리 상태 전이를 더 명확히 표현하기 위해 붙였다.

Options

Option 장점 문제 판단
same source면 항상 재계산 단순함 retry 때 불필요한 commit이 계속 생김 제외
corrected source를 append 구현 쉬움 같은 날짜 gold row가 중복될 수 있음 제외
whole-table overwrite 단순함 다른 날짜 partition까지 지울 위험 제외
business_date partition overwrite correction 범위가 명확함 Spark/Iceberg 설정과 test가 필요 선택
MERGE/upsert 강력함 이번 skeleton에 과함 backlog

이번 구현은 DataFrameWriterV2.overwritePartitions()를 사용했다.

corrected_df.writeTo("local.db.gold_daily_metrics").overwritePartitions()
Enter fullscreen mode Exit fullscreen mode

SQL INSERT OVERWRITE를 바로 쓰지 않은 이유는, 설정을 잘못 잡으면 static overwrite처럼 동작해 전체 table을 덮는 실수를 놓칠 수 있기 때문이다.

따라서 핵심 test는 단순히 "정정 날짜가 바뀌었나"가 아니라 이것이다.

D partition은 corrected rows로 교체된다.
D2 partition은 그대로 남는다.
Enter fullscreen mode Exit fullscreen mode

Decision

이번 slice는 local Spark/Iceberg walking skeleton으로 고정했다.

버전 pin:

Python: 3.10.12
Java: OpenJDK 17.0.19
PySpark: 3.5.8
Iceberg: 1.11.0
runtime jar: org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.11.0
catalog: local hadoop catalog
Enter fullscreen mode Exit fullscreen mode

구현 범위:

1. local SparkSession 생성
2. Iceberg hadoop catalog 설정
3. local.db.gold_daily_metrics table 생성
4. initial rows append
5. same source_hash retry는 skip처럼 처리해서 새 snapshot 없음
6. corrected rows로 business_date partition overwrite
7. current table rows 확인
8. snapshots metadata 확인
9. run_id -> snapshot_id evidence JSON 저장
Enter fullscreen mode Exit fullscreen mode

테이블은 하나다.

local.db.gold_daily_metrics
Enter fullscreen mode Exit fullscreen mode

partition column도 하나다.

business_date
Enter fullscreen mode Exit fullscreen mode

Evidence

구현 evidence:

requirements-spark.txt
src/manufacturing_data_platform/pipeline/spark_iceberg_skeleton.py
tests/test_spark_iceberg_skeleton.py
Enter fullscreen mode Exit fullscreen mode

검증 로그:

2026-07-11 — Spark/Iceberg single-gold-table walking skeleton
pytest tests/test_spark_iceberg_skeleton.py -q: 2 passed
pytest: 40 passed
Spark/Iceberg CLI: passed
Enter fullscreen mode Exit fullscreen mode

실행 명령:

pip install -r requirements-spark.txt

PYTHONPATH=src python -m manufacturing_data_platform.pipeline.spark_iceberg_skeleton \
  --warehouse /tmp/manufacturing-mini-iceberg-warehouse \
  --output-dir /tmp/manufacturing-mini-iceberg-evidence \
  --clean
Enter fullscreen mode Exit fullscreen mode

실제 evidence 일부:

snapshot_id 값은 Iceberg commit 때 생성되는 값이라 실행마다 달라질 수 있다. 중요한 것은 id 숫자 자체가 아니라, same-source retry에서는 새 snapshot이 없고 correction에서는 새 snapshot이 생긴다는 상태 변화다.

{
  "dataset_id": "manufacturing_daily_metrics",
  "table": "local.db.gold_daily_metrics",
  "business_date": "2026-06-29",
  "runs": [
    {
      "run_id": "spark-skeleton-r1",
      "source_hash": "source-hash-initial-001",
      "status": "processed",
      "gold_snapshot_id": 2920694863405545739,
      "snapshot_count": 1
    },
    {
      "run_id": "spark-skeleton-r1-retry",
      "source_hash": "source-hash-initial-001",
      "status": "skipped",
      "gold_snapshot_id": 2920694863405545739,
      "snapshot_count": 1
    },
    {
      "run_id": "spark-skeleton-r2",
      "source_hash": "source-hash-corrected-002",
      "status": "processed",
      "gold_snapshot_id": 8586499016384598474,
      "snapshot_count": 2
    }
  ],
  "partition_overwrite_assertions": {
    "target_partition_row_count": 1,
    "corrected_row_count": 1,
    "snapshot_increment": 1,
    "same_source_created_snapshot": false
  }
}
Enter fullscreen mode Exit fullscreen mode

current gold rows도 확인했다.

{
  "rows": [
    {
      "business_date": "2026-06-29",
      "plant_id": "plant-a",
      "line_id": "line-1",
      "product_code": "gearbox-a",
      "units_produced": 150,
      "defect_count": 6,
      "defect_rate": 0.04
    },
    {
      "business_date": "2026-06-30",
      "plant_id": "plant-a",
      "line_id": "line-1",
      "product_code": "gearbox-a",
      "units_produced": 50,
      "defect_count": 1,
      "defect_rate": 0.02
    }
  ]
}
Enter fullscreen mode Exit fullscreen mode

이 결과가 의미하는 것은 단순하다.

  • 2026-06-29는 corrected row 하나만 남았다.
  • 2026-06-30 partition은 사라지지 않았다.
  • 같은 source_hash retry는 새 snapshot을 만들지 않았다.
  • corrected source는 새 snapshot을 만들었다.
  • run_id와 snapshot_id를 구분해서 evidence로 남겼다.

Why Snapshot ID Is Not Run ID

run_id와 snapshot_id는 다른 세계의 id다.

run_id:
  pipeline execution identity
  예: spark-skeleton-r2

snapshot_id:
  Iceberg table commit identity
  예: 8586499016384598474
Enter fullscreen mode Exit fullscreen mode

이번 skeleton에서는 한 run이 gold table commit 하나를 만든다는 invariant를 뒀다.

one pipeline run -> one gold table commit -> one gold_snapshot_id
Enter fullscreen mode Exit fullscreen mode

그래서 run_id -> snapshot_id mapping이 가능하다.

나중에 한 run이 여러 Iceberg table에 commit하거나, 한 table에 여러 번 commit하면 이 관계는 1:N으로 바뀐다. 그건 backlog다.

Limitations

이건 production lakehouse가 아니다.

명확한 한계:

single gold table walking skeleton
full bronze/silver/gold Spark rewrite 아님
Spark-based quality suite 아님
MERGE/upsert 아님
Iceberg rollback system 아님
time-travel read demo는 아직 본문 claim으로 쓰지 않음
retention/expire snapshots 아님
concurrent writer handling 아님
Airflow-triggered Spark runtime 아님
cluster/Kubernetes runtime 아님
Enter fullscreen mode Exit fullscreen mode

그리고 이 구현은 local proof다.

local SparkSession
local hadoop catalog
/tmp warehouse
synthetic data
Enter fullscreen mode Exit fullscreen mode

이 범위를 넘겨서 "운영 lakehouse를 구축했다"고 말하면 과장이다.

Reference Links

이번 구현에서 참고한 공식 기준:

정리

same-source retry와 corrected-source rerun은 다르게 다뤄야 한다.

same source_hash
  -> skip
  -> no new snapshot

different source_hash + same business_date
  -> correction
  -> business_date partition overwrite
  -> new snapshot evidence
Enter fullscreen mode Exit fullscreen mode

이 작은 skeleton은 Spark/Iceberg 전체 이식이 아니라, 이 상태 전이가 실제로 가능한지 검증한 것이다.

코드: github.com/junhyun-dev/manufacturing-data-platform-mini

정리

source_hash skip만으로는 정정 파일을 설명하기 어렵다. 같은 입력은 skip해야 하지만, 같은 business_date의 다른 입력은 기존 결과를 중복 없이 교체해야 한다.

이 slice에서는 단일 local Iceberg gold table에서 그 상태 전이를 검증했다. 같은 source_hash rerun은 새 snapshot을 만들지 않고, 다른 source_hash correction은 DataFrameWriterV2.overwritePartitions()로 해당 business_date partition만 교체한다. 다른 날짜 partition 보존, 중복 row 부재, run_id와 snapshot_id 분리를 테스트로 확인했다.

범위는 명확하다. 이것은 production lakehouse나 full Spark medallion rewrite가 아니라, correction rerun을 Iceberg partition overwrite로 표현할 수 있는지 확인한 local walking skeleton이다.

Top comments (0)