sulco

메시지는 어떻게 쓰이고,
어떻게 다시 읽히나

Kafka의 쓰기 경로와, Flink가 offset을 믿는 법
0912 「압축은 왜 되는가」의 실전편
계기

파이프라인을 설계하다가 "왜?"에 답을 못 했다

브라우저 gateway Kafka Flink BigQuery 압축은 어디서? partition 몇 개? zone 넘는 비용은? 재시작하면 어디부터? 중복은 얼마나?
웹 성능 이벤트 수집 파이프라인 설계 중
오늘의 질문

질문 두 개로 줄이면 이렇다

1부 · 쓰기
메시지 한 건이 Kafka 디스크에 닿기까지, 압축은 어디서 걸리고 "썼다"는 무슨 뜻인가
2부 · 읽기
Flink가 Kafka에서 읽다가 죽으면 어디부터 다시 읽고, 그래서 중복은 어디서 생기나
1부

메시지 한 건이
Kafka 디스크에 닿기까지

압축은 메시지가 아니라 묶음에 걸린다
1부 · 기초 ①

보내는 쪽과 받는 쪽 사이에 Kafka를 두는 이유

직접 연결하면 gateway BigQuery 받는 쪽이 느려지면 보내는 쪽도 멈춘다 가운데 Kafka를 두면 gateway Kafka Flink 각자 자기 속도로 쓰고, 자기 속도로 읽는다
속도 차이를 흡수한다
받는 쪽이 잠깐 죽거나 느려져도 보내는 쪽은 계속 쓴다. 쌓인 건 나중에 따라잡는다
여러 곳이 같은 데이터를 읽는다
적재 job, 모니터링, 실험이 같은 메시지를 각자 읽는다
다시 읽을 수 있다
읽어도 지워지지 않는다. 2부의 핵심이 이 성질이다
1부 · 기초 ②

Kafka는 "꺼내면 사라지는 큐"가 아니라 "읽어도 남는 log"다

일반적인 큐 Kafka의 log CDE A·0B·1C·2D·3E·4 A, B는 꺼내는 순간 사라졌다 적재 job: 다음은 offset 2 모니터링: 다음은 offset 4 읽는 쪽마다 자기 위치(offset)를 따로 기억한다. 메시지는 보존 기간이 지나야 지워진다.
1부 · 기초 ③

용어 지도: cluster, broker, topic, partition, consumer group

cluster (서버 여러 대의 묶음) broker 1 broker 2 broker 3 events · P0 events · P1 events · P2 events · P1 사본 events · P2 사본 events · P0 사본 producer consumer group consumer 1consumer 2consumer 3 topic: 메시지 종류의 이름(events). partition: topic을 나눈 조각, broker들에 흩어져 있다. 사본(replica): 다른 broker에 둔 복제본. consumer group: 함께 나눠 읽는 consumer 묶음.
1부 · 기초 ④

메시지는 key와 value, 그리고 key가 partition을 정한다

# 메시지 한 건
key       "user-123"  (없어도 된다)
value     {"event_name": ..., "url": ...}
timestamp 1760000000000
headers   content-encoding=..., platform=...

# 어느 partition으로 가나
key 있음  → hash(key) % partition 수
key 없음  → 적당히 고르게 흩뿌림
key가 같으면 같은 partition
그래서 같은 사용자의 이벤트끼리는 순서가 지켜진다
우리 이벤트는 key가 없다
순서가 필요 없어서 흩뿌린다. 그래서 partition 수를 바꿔도 깨질 순서가 없다
1부 · 저장 단위

topic은 이름이고, 실체는 partition마다 있는 log다

partition 0partition 1partition 2 012345 0123 01234 뒤에 붙이기만 한다 (append-only) 칸 안의 숫자가 offset. 순서는 partition 안에서만 보장된다.
log는 segment 파일로 나뉘어 저장되고, 보존 기간이 지나면 segment째 지워진다
1부 · partition 수

partition 수가 곧 읽기 병렬도의 상한이다

partition 0 partition 1 partition 2 consumer 1 consumer 2 consumer 3 consumer 4 consumer 5 할 일이 없다 같은 group 안에서partition 하나 = consumer 하나
partition은 늘릴 수는 있어도 줄일 수 없다. 늘리면 key 기반 순서가 깨진다. 그래서 작게 시작했다(3개).
1부 · producer

send()는 바로 보내지 않는다. 묶어서 보낸다

msg msg msg msg partition별 버퍼 batch.size 만큼 차거나 linger.ms 만큼 기다리면 record batch 1개 요청 한 번으로 broker에 batch.size 기본 16KB · linger.ms 기본 0ms(Java 3.x) → 5ms(4.0, KIP-1030), librdkafka 5ms
1부 · record batch

batch는 header 하나와 record 여러 개, codec은 header에

$ kafka-dump-log.sh --files .../00000000000000000000.log
# 묶어서 보낸 topic
baseOffset: 0 lastOffset: 16 count: 17 ... size: 3771 magic: 2 compresscodec: zstd
# 한 건씩 보낸 topic
baseOffset: 0 lastOffset: 0 count: 1 ... size: 1025 magic: 2 compresscodec: zstd
record는 header 기준 상대 offset과 상대 시각만 들고 있어서 작다. 묶으면 17건이 3,771 bytes, 한 건은 1,025 bytes.
1부 · 압축

압축은 메시지마다가 아니라 batch에 한 번 걸린다

{"event_name":"...","platform":"desktop",
 "url":"https://...","sdk_version":"1.4.2", ...}
{"event_name":"...","platform":"desktop",
 "url":"https://...","sdk_version":"1.4.2", ...}
{"event_name":"...","platform":"desktop",
 ...
0912에서 본 것
압축은 반복을 참조로 바꾸는 일이다. 반복이 없으면 줄지 않는다.
그래서
JSON 이벤트는 같은 key를 매번 반복한다. 한 건이면 그 안의 반복만, 묶으면 건 사이의 반복까지 쓴다.
1부 · 실험 1

같은 이벤트도 묶을수록 더 줄어든다

묶은 건수원본 bytesgzipzstdlz4snappy
11,6672.2배2.2배1.6배1.6배
58,3435.2배5.1배3.5배3.4배
1016,7347.1배7.0배4.6배4.3배
3558,4109.9배10.0배6.2배5.6배
웹 성능 이벤트 35건(JSON, 건당 평균 1,667 bytes)을 n건씩 묶어 압축한 평균
1부 · 실험 2

Kafka에 실제로 2,000건을 넣고 segment 크기를 쟀다

codec한 건씩 flush (batch 2,000개)linger.ms 100 (batch 5개)차이
none3,614,8533,494,840−3%
gzip1,653,208309,129−81%
snappy2,197,827617,225−72%
lz42,215,371551,776−75%
zstd1,717,363114,726−93%
apache/kafka:3.9.0 한 대. 35건을 돌려 쓰되 id와 시각은 건마다 바꿨다. 단위는 bytes.
1부 · 실험에서 읽은 것

codec을 고르기 전에 batch부터 키워야 한다

약 2배
한 건씩 보내면 어떤 codec이든 이 정도에 머문다
3%
무압축이어도 한 건씩 보내면 batch header 비용이 붙는다
30배 대 11배
묶었을 때 zstd가 gzip을 크게 이겼다. 단, 돌려 쓴 데이터라 반복이 과장됐다. gzip은 32KB 안만, zstd는 훨씬 멀리 본다
1부 · broker

broker는 압축을 풀어 다시 묶지 않는다

producer묶고 압축한다 broker받은 batch 그대로 저장 consumer받아서 직접 푼다 압축된 bytes같은 bytes topic compression.type=producer(기본값)일 때 압축 비용은 양 끝이 내고, 디스크와 네트워크는 압축된 크기로 쓴다
topic codec을 producer와 다르게 지정하면 broker가 풀어서 다시 압축한다(broker CPU를 쓴다)
1부 · 설계로 돌아오면

브라우저의 gzip을 그대로 넘길까, 풀어서 다시 묶을까

그대로 넘기기
요청 하나가 batch 하나. 압축률은 나쁘지 않다. 하지만 페이지를 떠날 때 쓰는 sendBeacon은 header를 못 붙여 평문으로 오고, gateway는 내용을 못 보니 필수 field를 검증할 수 없다
풀어서 검증하고 producer가 묶기
여러 요청의 이벤트가 한 batch로 모여 브라우저 한 요청보다 큰 묶음이 된다. 압축률 손해가 없다. 조건은 하나, 이벤트마다 동기로 보내면 batch가 작아진다
1부 · "썼다"의 뜻

producer는 leader에게 쓰고, acks가 성공 시점을 정한다

acks성공으로 보는 시점잃을 수 있는 경우
0보내자마자무엇이든
1leader가 자기 log에 썼을 때follower가 따라오기 전에 leader가 죽으면
allISR 전체가 썼을 때 (3.0부터 기본값)ISR이 min.insync.replicas보다 작으면 쓰기를 거절해 잃지 않는다
ISR: leader를 충분히 따라온 replica 집합. consumer는 ISR 전체가 가진 지점(high watermark)까지만 읽는다.
1부 · 비용

leader에만 쓰니 zone을 넘고, 넘으면 돈이 든다

zone a zone b zone c producer leader follower follower consumer 과금 기본: leader에서 읽음 KIP-392 zone이 셋이면 쓰기의 약 2/3가 zone을 넘는다 · broker 사이 복제는 MSK가 과금하지 않는다
broker에 RackAwareReplicaSelector, consumer에 client.rack을 주면 가까운 follower에서 읽는다
1부 · 실제로 본 것

설정을 넣었다고 쓰이는 건 아니었다

# deployment
env:
  - name: KAFKA_CLIENT_RACK
    valueFrom: ... topology.kubernetes.io/zone

# source 코드
KafkaSource.builder()
  .setBootstrapServers(...)
  .setGroupId(...)
  // client.rack 설정 없음
참고하던 job
zone 값을 env로 넣어 놓고 코드가 읽지 않아 rack-aware가 아니었다
새 job
zone을 읽어 client.rack에 넣는 코드가 있는 템플릿 쪽 방식을 쓴다
2부

Flink는 Kafka offset을
어떻게 믿나

죽었다 살아나면 어디부터 다시 읽고, 무엇이 두 번 들어가나
2부 · 기초 ①

Flink는 끝이 없는 데이터를 계속 처리하는 엔진이다

배치 처리 (Spark job, SQL 쿼리)
"어제 하루치"처럼 끝이 있는 데이터를 한 번 읽고 끝낸다. 실패하면 처음부터 다시 돌리면 된다
스트림 처리 (Flink)
Kafka처럼 계속 들어오는 데이터를 들어오는 대로 처리하고, 멈추지 않는다. 실패했을 때 처음부터 다시 돌릴 수가 없다
"처음부터 못 돌린다"가 2부 전체의 문제다. 그럼 어디서부터 다시 하나?
2부 · 기초 ②

Flink job은 source에서 sink까지 흐르는 그래프다

sourceKafka에서 읽기 operator풀기, 검증, 변환 sinkBigQuery에 쓰기 검증 실패는 error topic으로 (두 번째 sink) 레코드 하나하나가 이 길을 따라 흐른다
2부 · 기초 ③

각 단계는 여러 벌로 복제돼 TaskManager에서 동시에 돈다

JobManager · 조율, checkpoint 지시 TaskManager 1 (pod) TaskManager 2 (pod) slot slot slot slot (빈자리) source #1 ← P0operator #1sink #1 source #2 ← P1operator #2sink #2 source #3 ← P2operator #3sink #3 parallelism 3 = 단계마다 subtask 3개. source subtask가 partition을 하나씩 맡는다. 그래서 partition 3개와 parallelism 3을 맞춘다
2부 · 기초 ④

operator가 기억하는 값이 state고, source의 state는 offset이다

state의 예
"지난 5분 동안 몇 건", "이 사용자의 마지막 이벤트"처럼 레코드 사이를 이어 주는 값
source의 state
partition마다 어디까지 읽었나. 우리 job처럼 집계가 없어도 이것만은 있다
문제
state는 TaskManager 메모리에 있다. pod가 죽으면 같이 사라진다. 그래서 주기적으로 밖(S3)에 저장한다. 이것이 checkpoint
2부 · 전달 보장

보장 수준은 "offset을 언제 저장하느냐"로 갈린다

at-most-once
읽자마자 위치를 저장하고 처리한다. 처리 중에 죽으면 그 건은 다시 안 읽는다. 유실은 있고 중복은 없다
at-least-once
처리하고 나서 위치를 저장한다. 저장 전에 죽으면 또 처리한다. 유실은 없고 중복은 있다
exactly-once
위치 저장과 결과 쓰기를 한 덩어리로 확정한다. 둘 다 되거나 둘 다 안 된다
일반 Kafka consumer의 기본값 enable.auto.commit은 5초마다 위치를 저장한다. 처리와 상관없이 저장해서 둘 다 생길 수 있다
2부 · 전달 보장

같은 장애, 세 가지 결과

100120150160 마지막 checkpoint 여기서 죽음 120~150, 30건 at-most-once위치는 이미 150으로 저장됐다. 150부터 읽으니 처리 중이던 건이 사라진다 at-least-once120부터 다시 읽는다. sink에 이미 쓴 30건이 한 번 더 들어간다 exactly-once120부터 다시 처리하는 건 같다. 30건의 쓰기가 확정 전이라 버려져서 한 번만 남는다
exactly-once는 "한 번만 처리"가 아니라 "여러 번 처리해도 결과는 한 번만 남는다"는 뜻이다
2부 · offset

Flink는 Kafka에 commit한 offset을 믿지 않는다

보통의 consumer
읽은 위치를 Kafka 내부 topic __consumer_offsets에 commit하고, 다시 시작하면 거기서 읽는다
Flink Kafka source
읽은 offset을 자기 checkpoint에 넣고, 복구하면 checkpoint의 offset부터 읽는다. Kafka commit은 checkpoint가 끝난 뒤에 하고, lag 모니터링용일 뿐이다
이유: offset과 연산자 상태가 "같은 순간"의 것이어야 한다. 따로 저장하면 둘이 어긋나는 시점이 생긴다.
2부 · consumer lag

lag는 "쌓였는데 아직 못 읽은 양"이다

012345678 committed offset = 5 log end offset = 9 lag = 9 − 5 = 4 lag가 계속 늘면읽는 쪽이 못 따라가는 중 Flink는 checkpoint가 끝날 때만 Kafka에 commit한다. 그래서 Kafka에서 본 lag는 checkpoint 주기(60초)마다 톱니처럼 오르내린다. 실제로 늦은 게 아니다.
실시간 지연은 Flink가 내보내는 Kafka source 지표(records-lag-max 등)로 본다
2부 · checkpoint

barrier가 데이터 사이를 흘러 "같은 순간"을 만든다

sourceoffset 저장 연산자상태 저장 sink도착하면 완료 barrier n barrier n barrier 앞은 전부 상태에 반영됐고, 뒤는 하나도 반영되지 않은 순간 복구하면 저장한 offset부터 다시 읽는다. Kafka가 다시 읽기를 지원하니까 성립한다.
2부 · at-least-once

Flink 안에서 중복은 두 곳에서 생기고, 우리 job은 거의 sink 쪽이다

① barrier를 정렬하지 않는다
AT_LEAST_ONCE 모드는 다른 채널의 barrier를 기다리지 않는다. barrier 뒤의 레코드가 상태에 섞일 수 있다. 상태 없는 job에는 거의 영향이 없다
② sink가 이미 쓴 것
마지막 checkpoint 뒤에 BigQuery로 쓴 레코드는 되돌릴 수 없다. offset을 되감으면 또 쓴다. 크기는 최대 checkpoint 주기만큼(60초)
2부 · producer 쪽 중복

재시도가 중복을 만들고, idempotent producer가 막는다

# idempotence 끔 (1부 실험에 쓴 client의 기본값)
baseOffset: 0 count: 3 baseSequence: -1 lastSequence: -1 producerId: -1
# idempotence 켬
baseOffset: 0 count: 3 baseSequence: 0 lastSequence: 2 producerId: 0
baseOffset: 3 count: 3 baseSequence: 3 lastSequence: 5 producerId: 0
어떻게 막나
producer마다 id를, batch마다 순번을 붙인다. broker는 이미 받은 순번이 또 오면 버린다. 응답을 못 받아 재시도해도 한 번만 저장된다
어디까지 막나
Java client는 3.0부터 기본으로 켜져 있다. 단, producer가 재시작하면 id가 바뀌어 그 전에 보낸 것과는 구별하지 못한다
2부 · exactly-once

끝까지 한 번만 쓰려면 sink가 둘 중 하나여야 한다

멱등 쓰기
같은 레코드를 두 번 써도 결과가 같다. 고정 key로 upsert한다
트랜잭션 쓰기 (2PC)
checkpoint 때 미리 준비해 두고, 완료 알림을 받으면 commit한다. Flink Kafka sink가 이렇게 한다. 대가는 최대 checkpoint 주기만큼의 지연
checkpoint를 EXACTLY_ONCE로 바꾸는 것만으로는 Flink 안쪽 상태까지만 보장된다. 2PC에서 Kafka가 실제로 하는 일이 다음 장이다
2부 · Kafka transaction

abort된 메시지도 log에 남는다. 읽는 쪽이 건너뛸 뿐이다

# commit 3건 → abort 2건 → commit 2건
offset 0-2  data    isTransactional: true
offset 3    control endTxnMarker: COMMIT
offset 4-5  data    isTransactional: true
offset 6    control endTxnMarker: ABORT
offset 7-8  data    isTransactional: true
offset 9    control endTxnMarker: COMMIT
isolation.level읽은 offset
read_uncommitted0 1 2 4 5 7 8
read_committed0 1 2 7 8
control record(3, 6, 9)는 어느 쪽에도 보이지 않는다. offset이 중간중간 비는 이유다.
로컬 Kafka 3.9.0에서 kafka-dump-log와 consumer 두 개로 확인
2부 · checkpoint와 savepoint

checkpoint는 장애 복구용, savepoint는 배포용이다

checkpointsavepoint
누가 찍나Flink가 주기적으로 자동사람이나 배포 도구가 명령해서
언제 쓰나장애가 나서 job이 스스로 다시 뜰 때코드를 바꿔 다시 배포할 때, 버전을 올릴 때, 옮길 때
얼마나 남나최근 것만 남기고 Flink가 지운다사람이 지울 때까지 남는다
담긴 것둘 다 같다: 각 operator의 state와 source의 offset
배포할 때 savepoint를 찍고 그 지점부터 새 버전을 띄우면, 배포 전후로 빠지거나 겹치는 offset이 없다. 코드를 바꿔도 state를 찾아가도록 operator마다 고정 uid를 붙여 둔다.
2부 · 설계로 돌아오면

그래서 dedup 없이 at-least-once로 간다

at-least-once, 60초
상태가 없는 job이라 barrier 정렬로 얻을 것이 없다
dedup 안 함
성능 이벤트는 백분위로 본다. 장애 때 최대 1분치가 두 번 들어가도 분포는 거의 그대로다. 이벤트마다 id가 있어 조회 때 거를 수도 있다
savepoint로 배포
savepoint에 offset이 들어 있어 배포 전후로 빠지거나 겹치는 구간이 없다
2부 · 안전장치

중복이 들어와도 조회할 때 id로 거를 수 있다

-- 이벤트마다 있는 id로 한 건만 남긴다
SELECT *
FROM events
WHERE dt = '2026-10-10'
QUALIFY ROW_NUMBER() OVER (PARTITION BY id ORDER BY received_at) = 1
적재 단계에서 거르기
테이블은 늘 깨끗하지만, 매번 이미 쓴 id와 비교하는 비용이 들고 파이프라인이 복잡해진다
조회 단계에서 거르기
중복이 드물고 집계 위주라면 필요한 쿼리에서만 거르는 쪽이 싸다. 우리는 이쪽을 골랐다
마무리

압축은 묶음에서 이기고,
정확성은 sink에서 결정된다

"codec보다 batch, checkpoint보다 sink를 먼저 본다."
0912 압축 편의 "반복이 있어야 줄어든다"가 Kafka에서는 "묶어야 줄어든다"였다