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

이번 구현은 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-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)

이번 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

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

• Apache Spark 3.5.8 docs
• Apache Iceberg Spark Getting Started
• Apache Iceberg Multi-Engine Support
• Maven Central: iceberg-spark-runtime-3.5_2.12

정리

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이다.