skip에서 partition overwrite로: business_date 재처리를 Iceberg로 다시 표현하기
이전 글에서는 같은 source_hash가 다시 들어왔을 때 기존 successful run을 재사용하는 idempotency를 다뤘다.
하지만 재처리에는 두 종류가 있다.
1. 같은 입력이 다시 들어온 경우
-> skip이 맞다.
2. 같은 business_date의 정정 입력이 들어온 경우
-> skip하면 안 된다.
-> 같은 날짜의 gold 결과를 중복 없이 교체해야 한다.
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
나중에 같은 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가 남음
그래서 이 slice의 질문은 이렇다.
같은 business_date의 정정 source를 처리할 때,
gold table에서 해당 날짜 partition만 중복 없이 교체하고,
어떤 run이 어떤 Iceberg snapshot을 만들었는지 남길 수 있는가?
Decision Pressure
Slice1의 CSV pipeline은 already-successful source를 안전하게 skip할 수 있다.
dataset_id + business_date + source_hash
이 key가 같으면 같은 입력이다. 다시 계산해도 같은 결과이므로 기존 run을 재사용한다.
하지만 source_hash가 달라졌다면 의미가 다르다.
same business_date
different source_hash
이건 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
즉 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()
SQL INSERT OVERWRITE를 바로 쓰지 않은 이유는, 설정을 잘못 잡으면 static overwrite처럼 동작해 전체 table을 덮는 실수를 놓칠 수 있기 때문이다.
따라서 핵심 test는 단순히 "정정 날짜가 바뀌었나"가 아니라 이것이다.
D partition은 corrected rows로 교체된다.
D2 partition은 그대로 남는다.
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
구현 범위:
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 저장
테이블은 하나다.
local.db.gold_daily_metrics
partition column도 하나다.
business_date
Evidence
구현 evidence:
requirements-spark.txt
src/manufacturing_data_platform/pipeline/spark_iceberg_skeleton.py
tests/test_spark_iceberg_skeleton.py
검증 로그:
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
실행 명령:
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
실제 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
}
}
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
}
]
}
이 결과가 의미하는 것은 단순하다.
2026-06-29는 corrected row 하나만 남았다.
2026-06-30partition은 사라지지 않았다.- 같은
source_hashretry는 새 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
이번 skeleton에서는 한 run이 gold table commit 하나를 만든다는 invariant를 뒀다.
one pipeline run -> one gold table commit -> one gold_snapshot_id
그래서 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 아님
그리고 이 구현은 local proof다.
local SparkSession
local hadoop catalog
/tmp warehouse
synthetic data
이 범위를 넘겨서 "운영 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
이 작은 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이다.
SOCIAL SHARE CARD GENERATOR