source_hash로 같은 입력 재처리를 안전하게 skip하기

Scenario

같은 business_date의 제조/로봇 이벤트 파일을 다시 처리해야 하는 상황이 있다. 예:

• retry: 앞 run이 중간에 실패해서 다시 실행한다.
• backfill: 과거 날짜를 다시 채운다.
• operator mistake: 같은 파일을 실수로 다시 실행한다.

단순히 매번 append하면 같은 날짜의 gold metric이 중복될 수 있다.

Decision Pressure

단순 CSV pipeline은 보통 이렇게 끝난다.

CSV 읽기 -> silver 만들기 -> gold 집계 -> 결과 저장

하지만 운영 관점에서는 질문이 생긴다.

• 이 입력은 전에 처리한 파일과 같은가?
• 같은 파일을 다시 돌리면 중복 output이 생기나?
• 다른 파일로 같은 날짜를 다시 돌리면 어떻게 구분하나?
• 어떤 run이 어떤 source에서 만들어졌나?

그래서 재실행을 판단할 identity가 필요했다.

Options

이 프로젝트의 Slice1은 마지막 선택지를 쓴다.

Decision

현재 mini pipeline은 입력 파일의 content hash를 source_hash로 계산한다.

• idempotency key: dataset_id + business_date + source_hash
• 이미 성공한 run이 있으면 새로 처리하지 않고 기존 run을 재사용한다.
• same dataset_id same business_date same source_hash prior successful run exists => status = skipped

이 선택은 작지만 중요하다.

• 같은 파일 재실행: skip -> 중복 없음
• 같은 날짜의 정정 파일: source_hash가 다름 -> skip하지 않고 새 run으로 처리 가능

단, 여기서 조심해야 할 경계가 있다. Slice1은 다른 source_hash를 새 run으로 처리할 수 있지만, 이전 gold partition을 원자적으로 교체하는 Iceberg-style overwrite까지 구현한 것은 아니다. 그 문제는 다음 Slice2의 business_date partition overwrite 주제다.

Evidence

관련 코드와 검증 evidence:

• src/manufacturing_data_platform/pipeline/lakehouse.py
• tests/test_lakehouse_pipeline.py
• VERIFICATION_LOG.md
• README.md

검증 로그 (2026-07-08 publication readiness check):

• pytest: 33 passed
• lakehouse JSON CLI: passed, status=processed, quality_passed=true
• EAV JSON CLI: passed, status=processed, quality_passed=true

이 repo의 테스트는 같은 입력 재실행 시 status="skipped"가 되는 경로를 구체적으로 확인한다.

• tests/test_lakehouse_pipeline.py
  • test_rerun_same_source_and_date_is_skipped_mongo
  • test_rerun_same_source_and_date_is_skipped_json

Mongo backend 쪽 테스트는 재사용 감사 흔적인 reuse_count가 증가하는지도 확인한다. 즉 단순히 "두 번째 실행을 무시"하는 것이 아니라, 같은 성공 run을 재사용했다는 기록을 남긴다. 또한 JSON catalog backend로 Mongo 없이도 CLI smoke run을 검증했다.

Limitations

정직한 한계:

• 이건 production lakehouse가 아니다.
• Spark/Iceberg는 아직 구현되지 않았다.
• pyspark도 현재 repo 환경에 설치되어 있지 않다.
• Kafka streaming도 없다.
• real Mongo runtime verification은 환경 제약으로 미검증이다.
• Airflow runtime trigger도 미검증이다.

현재 claim은 이렇게 제한한다: synthetic CSV 기반 mini pipeline에서 source_hash를 이용한 idempotent rerun을 구현하고 테스트/CLI로 검증했다.

Why This Connects To Iceberg Later

Slice1의 skip 전략은 같은 입력 재실행을 안전하게 막는다. 하지만 다음 질문이 남는다.

• 정정된 파일로 같은 business_date를 다시 처리해야 한다면?
• append하면 중복된다.
• skip하면 정정을 반영하지 못한다.
• overwrite하면 어디까지 덮어써야 하는가?

이 질문이 Slice2의 Spark/Iceberg 주제로 이어진다.

• same source_hash: skip
• different source_hash for same business_date: 현재 Slice1에서는 새 run으로 처리 가능
• 중복 없는 partition 교체는 Iceberg partition atomic overwrite 후보

그래서 다음 글 후보는: skip에서 partition overwrite로: business_date 재처리를 Iceberg로 다시 표현하기

정리

같은 파일을 다시 처리하는 retry와, 같은 날짜의 정정 파일을 다시 처리하는 correction은 다른 문제다. 이 slice에서는 dataset_id + business_date + source_hash로 같은 입력을 skip하고, reuse_count로 재사용 흔적을 남겼다. 정정 파일을 중복 없이 교체하는 문제는 아직 이 글의 구현 범위가 아니며, 다음 Spark/Iceberg slice에서 business_date partition overwrite로 다룬다.