DEV Community

박준현
박준현

Posted on

Kafka event를 바로 Spark로 보내지 않고 batch bridge를 둔 이유

#dataengineering #kafka #python #apacheiceberg

Kafka event를 바로 Spark로 보내지 않고 batch bridge를 둔 이유

Kafka를 붙이고 나면 다음 선택은 자연스럽게 보인다.

Kafka
-> Spark Structured Streaming
-> Iceberg
Enter fullscreen mode Exit fullscreen mode

하지만 이 프로젝트에서는 그렇게 하지 않았다.

먼저 Kafka event를 payload와 topic/partition/offset이 포함된 immutable JSONL로 보존했다. 그다음 한 business_date의 accepted event만 결정적 CSV로 바꾸는 작은 bridge를 두고, 이미 검증한 batch quality/gold/Iceberg 경로를 재사용했다.

이유는 간단하다.

이번 시나리오의 압력은 continuous window 처리가 아니라, event를 잃지 않고 replay 가능한 raw로 보존한 뒤 같은 accepted set이 trusted 결과를 두 배로 만들지 않게 하는 것이었다.

Scenario

합성 설비 event가 local Kafka topic으로 들어온다.

producer
-> Kafka topic
-> bounded raw consumer
-> immutable JSONL landing
Enter fullscreen mode Exit fullscreen mode

Kafka K1 and K1.5 runtime overview

consumer가 event를 읽었다는 사실과 downstream에 안전하게 보존했다는 사실은 같지 않다.

가장 신경 쓴 실패 구간은 이것이었다.

1. Kafka record poll
2. raw landing write 완료
3. offset commit 전에 consumer crash
4. 같은 record 재전달
Enter fullscreen mode Exit fullscreen mode

3번에서 죽으면 Kafka는 마지막 committed position부터 다시 전달한다. 이때 landing이 같은 record를 또 accepted로 쓰면 raw부터 중복된다.

Decision Pressure: 무엇을 먼저 확정할 것인가

Kafka offset commit과 local filesystem write는 하나의 transaction이 아니다.

선택지는 두 가지였다.

순서 장점 실패 시 문제
offset commit -> file write 중복 가능성이 작아 보임 write 실패 시 record를 잃을 수 있음
file write -> offset commit record loss를 피함 commit 전 crash 시 같은 record가 재전달됨

이 프로젝트는 두 번째를 선택했다.

poll bounded records
-> validate / classify
-> staging에 JSONL + manifest write
-> fsync
-> immutable batch directory로 rename
-> synchronous offset commit
Enter fullscreen mode Exit fullscreen mode

Kafka 공식 consumer 문서도 processing과 결합할 때 auto commit을 끄고 처리가 끝난 뒤 수동 commit하면, write 이후 commit 전 실패 구간에서 record가 재전달되는 at-least-once 특성이 생긴다고 설명한다.

코드의 경계도 그대로 드러난다.

landing = land_records(
    records,
    output_dir,
    simulate_crash_after_rename=simulate_crash_after_landing,
)

offsets = [
    TopicPartition(item["topic"], item["partition"], item["next_offset"])
    for item in landing.committable_offsets
]
consumer.commit(offsets=offsets, asynchronous=False)
Enter fullscreen mode Exit fullscreen mode

핵심은 재전달 자체를 없애는 것이 아니다.

at-least-once 재전달은 허용한다.
이미 durable한 (topic, partition, offset)은 다시 accepted하지 않는다.
Enter fullscreen mode Exit fullscreen mode

Failure Injection: 정말 복구되는가

정상 경로만 실행해서는 이 계약을 증명할 수 없다. 그래서 offset 3을 landing한 직후, commit 전에 의도적으로 예외를 발생시켰다.

Kafka landing-before-commit failure recovery

실제 검증 결과:

produced coordinates: 5
accepted events: 4
quarantined events: 1
persisted coordinates: 5

injected crash: observed
redelivered coordinate: 1
recovery landing status: reused
accepted total after recovery: 4
committed next offset: 4
Enter fullscreen mode Exit fullscreen mode

재전달은 일어났지만 accepted set은 늘지 않았다.

별도의 bounded replay도 실행했다.

replay offsets: 0..3
reused coordinates: 4
normal consumer-group commit: not performed
Enter fullscreen mode Exit fullscreen mode

replay가 정상 consumer-group progress를 바꾸지 않는 것도 별도 계약으로 확인했다.

왜 바로 Spark Structured Streaming으로 가지 않았나

K1 raw landing이 끝나도 기존 pipeline과는 입력 형식이 다르다.

K1 output: accepted JSONL + manifest
기존 pipeline input: one business_date CSV
Enter fullscreen mode Exit fullscreen mode

여기서 세 가지 선택지를 비교했다.

Option 장점 이번 slice의 비용 판단
Spark Structured Streaming window, checkpoint, continuous 처리 가능 아직 없는 latency/window 요구까지 운영해야 함 Backlog
direct Kafka -> Iceberg sink 경로가 짧음 기존 quality/catalog gate를 우회 제외
deterministic batch bridge 기존 검증 자산 재사용 adapter contract가 하나 필요 선택

K1.5 bridge는 한 날짜의 accepted event를 (topic, partition, offset) 순서로 정렬하고, 고정 header와 newline으로 canonical CSV를 만든다.

여기서 결정적이라는 말은 주어진 하나의 immutable landing에서 같은 accepted set을 다시 선택했을 때 같은 bytes가 나온다는 뜻이다. producer를 새로 실행해 Kafka timestamp나 record fingerprint가 바뀌면 provenance가 달라지므로 새 source_hash가 만들어진다.

selected.sort(key=lambda item: item.sort_key)
csv_bytes = canonical_csv_bytes(selected)
source_hash = sha256(csv_bytes).hexdigest()
Enter fullscreen mode Exit fullscreen mode

이 CSV에는 business column뿐 아니라 다음 provenance도 들어간다.

event_id
schema_version
kafka_topic
kafka_partition
kafka_offset
kafka_key
kafka_timestamp_ms
kafka_record_fingerprint
Enter fullscreen mode Exit fullscreen mode

그리고 CSV의 SHA-256을 기존 pipeline의 source_hash로 그대로 사용한다.

same accepted Kafka set from one immutable landing
-> same canonical CSV bytes
-> same source_hash
-> same adapter version reused
-> existing lakehouse run skipped
-> Iceberg retry creates no new snapshot
Enter fullscreen mode Exit fullscreen mode

새 idempotency 체계를 하나 더 만든 것이 아니라, Kafka provenance를 기존 source identity 안으로 연결했다.

이 경로에는 서로 다른 다섯 개의 identity가 등장한다. 헷갈리면 안 되는 부분이다.

Identity 무엇을 가리키나 누가 정하나 같은 accepted set 재실행 시
(topic, partition, offset) Kafka log 안의 transport 위치 Kafka broker coordinate 그대로 재사용
event_id business event 한 건의 정체 producer가 생성 같은 event_id는 accepted set을 늘리지 않음
adapter source_hash 선택된 한 날짜 batch 입력의 정체 canonical CSV의 SHA-256 같은 값 → adapter version 재사용
pipeline run_id lakehouse 실행 한 번의 정체 pipeline이 실행 시 생성 재실행은 기존 run_id를 skip으로 재사용
Iceberg snapshot_id gold table commit 한 번의 정체 Iceberg가 write 시 생성 새 snapshot을 만들지 않음

핵심은 transport 위치(offset)와 business 정체(event_id)와 batch 정체(source_hash)를 분리해 두었기 때문에, 재전달이나 재실행이 어느 층에서 일어나도 그 위쪽 trusted 결과가 늘지 않는다는 것이다.

End-to-End Evidence

Kafka landing to batch and Iceberg rerun evidence

2026-07-16 local runtime 결과:

Kafka: 4.3.1, one broker / one topic / one partition
K1 broker verification: passed
K1.5 checks: 11 passed

selected accepted events: 4
adapter: created -> reused
lakehouse: processed -> skipped
quality checks: 8 / 8 pass
silver rows: 4
gold rows: 1
gold totals: units=100, defects=6

Iceberg: published -> skipped
snapshot count: 1 -> 1
snapshot ID unchanged: true
Enter fullscreen mode Exit fullscreen mode

또한 adapter는 요청 날짜에 accepted event가 하나도 없거나 manifest와 JSONL이 어긋나면 기존 pipeline을 호출하기 전에 실패한다. 그렇지 않으면 기존 pipeline의 synthetic sample 생성 경로가 실제 입력이 없는데도 성공처럼 보이는 false green을 만들 수 있기 때문이다.

What This Changed in the Design

처음에는 Kafka 다음에 Spark Structured Streaming을 붙이는 순서를 생각했다.

실제 시나리오와 failure contract를 따라가자 순서가 바뀌었다.

K1   Kafka -> immutable raw landing
K1.5 raw landing -> deterministic batch adapter -> existing quality/gold/Iceberg
K2   Spark Structured Streaming은 window/latency pressure가 생길 때만
Enter fullscreen mode Exit fullscreen mode

도구의 자연스러운 나열보다 현재 문제를 닫는 가장 작은 vertical slice를 선택한 것이다.

Limitations

이 글이 증명하는 범위:

local Kafka one broker / one topic / one partition
bounded producer and consumer
landing-before-commit failure injection
coordinate-based raw reuse
bounded replay without normal group commit
one business_date deterministic batch bridge
existing local quality/gold/Iceberg retry contract reuse
Enter fullscreen mode Exit fullscreen mode

증명하지 않은 범위:

continuous streaming service
multi-partition ordering / rebalance
multi-broker HA
Spark Structured Streaming
direct Kafka-to-Iceberg sink
end-to-end exactly-once
power-loss durability
production security / scale / operations
Enter fullscreen mode Exit fullscreen mode

특히 os.replace와 fsync 순서는 local Linux filesystem에서만 검증했다. in-process 예외를 주입한 것이므로 전원 손실까지 견디는 crash consistency를 주장하지 않는다.

정리

Kafka를 썼다는 사실보다 중요한 것은 언제 offset을 확정하고, 재전달을 어떤 identity로 흡수하며, 기존 trusted pipeline과 어떻게 연결하는가였다.

이번 구현의 핵심은 세 줄이다.

durable landing 전에 offset을 commit하지 않는다.
재전달된 Kafka coordinate는 accepted set을 늘리지 않는다.
같은 accepted set은 기존 source_hash 계약을 통해 gold와 snapshot도 늘리지 않는다.
Enter fullscreen mode Exit fullscreen mode

코드와 전체 walkthrough: manufacturing-data-platform-mini

Sources

Top comments (0)