.
지난 글에서 Flink on K8s의 아키텍처와 배포 모델을 정리했다. 이번엔 실제로 Kafka → 변환 → Redis 실시간 파이프라인을 운영하면서 이해하게 된 런타임 내부 동작을 정리한다.
Kafka 컨슈머와 Flink는 같은 토픽을 읽어도 분할·상태·오프셋을 다루는 모델이 근본적으로 다르다. 데이터가 파이프라인을 흐르며 각기 다른 기준으로 재분할되는 과정을 순서대로 따라간다.
입력 분할: Kafka Partition과 FLIP-27의 할당
일반 Kafka 컨슈머는 Consumer Group의 Rebalance로 파티션을 나눠 갖는다. 파티션 수가 곧 병렬성의 상한이고 컨슈머가 추가/이탈하면 브로커가 재분배를 트리거한다.
Flink의 FLIP-27 KafkaSource는 이 방식을 쓰지 않는다.
FLIP-27
Flink 1.11부터 도입된 Source API 스펙
기존 SourceFunction 기반 API가 파티션 발견, 워커 간 분배, 이벤트 타임 정렬 같은 걸 커넥터마다 각자 구현해야 했던 문제를 표준화하기 위해 나옴
•
SplitEnumerator: JobManager에서 동작. 소스의 분할 단위(Kafka면 파티션)를 발견하고, 어느 subtask에 할당할지 결정
•
SourceReader: TaskManager에서 동작. 할당받은 split을 실제로 읽어서 레코드를 만들어냄
이 표준화 덕분에 파티션 할당 로직을 Kafka의 Consumer Group 메커니즘에 기대지 않고 Flink가 직접 제어할 수 있게 됐다
•
JobManager의 SplitEnumerator가 각 파티션(split)을 source subtask에 직접 할당
•
Kafka의 Group Coordination에 참여하지 않음 → 파티션 분배 주체가 Kafka 브로커가 아니라 Flink 내부 로직
이 차이로 Kafka 파티션 수와 Flink parallelism은 독립적인 축이 된다.
•
파티션 1개 = source subtask 1개에만 할당 (나눠 갖지 않음)
•
따라서 source parallelism > 파티션 수가 되면 초과분 subtask는 유휴 상태
상태 분할: keyBy와 Key-Group
소비된 스트림은 keyBy()를 만나면서 Key-Group 이라는 단위로 다시 분할된다.
Keyed State는 key 단위로 유지되는데 Flink는 key를 subtask에 직접 매핑하지 않고 그 사이에 고정 개수의 논리적 버킷을 둔다.
keyGroupId = murmurHash(key.hashCode()) % maxParallelism
JavaScript
복사
•
maxParallelism이 곧 Key-Group의 총 개수
•
각 subtask는 여러 Key-Group을 연속된 범위(KeyGroupRange) 로 할당
인다이렉션 두는 이유 - rescale 비용 문제
key를 subtask에 직접 해시(% parallelism)한다고 가정하면:
•
parallelism이 4 → 16으로 바뀌는 순간, % 4와 % 16의 결과가 대부분의 key에서 달라짐
•
즉 거의 모든 key의 소속 subtask가 바뀌고, 이는 State 전체를 다시 분배해야 한다는 뜻
Key-Group을 고정해두면:
•
rescale은 이미 나뉜 128개 Key-Group을 몇 개 subtask에 다시 배분할지만 결정하면 됨
•
key 하나하나를 다시 해싱할 필요 없이 버킷(Key-Group) 단위로만 이동하면 됨
이 구조가 만드는 두 가지 제약
1.
maxParallelism은 최초 설정 후 변경 불가
•
Savepoint는 State를 Key-Group 단위로 스냅샷함
•
maxParallelism이 바뀌면 Key-Group 총 개수가 바뀌고 복원 시 매핑이 어긋남
2.
maxParallelism은 보통 2ⁿ으로 설정
•
2/4/8/16처럼 흔한 parallelism으로 나눌 때 Key-Group이 균등 분배됨
•
홀수나 소수로 두면 특정 subtask에 Key-Group이 쏠려 부하 편중 발생 가능
요약하자면,
•
입력 분할(Kafka Partition)과 상태 분할(Key-Group)은 서로 다른 해시 기준을 가진 완전히 별개의 층위
•
예를 들어 파티션이 15개인 토픽이어도 내부 State는 그와 무관하게 maxParallelism(예: 128)개의 Key-Group으로 재분배되어 흐름
오프셋 관리: Checkpoint에 묶이는 소비 위치
Flink는 소비 위치(offset)를 Kafka가 아니라 자기 자신의 Checkpoint로 관리한다.
•
offset은 Operator State로 저장됨
•
체크포인트 시점에 소스의 offset과 다운스트림 Keyed State가 원자적으로 함께 스냅샷됨
•
그래서 장애로 TaskManager가 재기동되어도 어디까지 읽었고 그 시점의 상태는 무엇이었는지가 항상 일관되게 복원됨 → 이게 exactly-once(또는 at-least-once) 보장의 근간
여기서 헷갈리기 쉬운 지점 - Kafka의 Committed Offset은 복구 근거가 아니다
•
Flink가 체크포인트 시점에 Kafka에도 offset을 커밋하지만 이건 모니터링용 best-effort 값일 뿐
•
실제 장애 복구는 언제나 Checkpoint state 기반
이 구조 때문에 lag 지표가 두 개로 갈린다:
지표 | 정의 | 성격 |
Committed offset 기반 lag | 토픽 최신 offset − committed offset | 체크포인트 주기에만 갱신 |
records_lag_max (flink 지표) | 토픽 최신 offset − 실제 읽는 position | 실제 소비 진행도를 반영 |
예를 들어 체크포인트 주기가 1분이면 Kafka 쪽 모니터링 툴이 보여주는 lag은 최대 1분치 지연을 안고 갱신된다. 실제 소비 지연을 보려면 records_lag_max를 봐야 한다.
Consumer Group의 역할
Flink는 group.id를 설정하지만 실제로 그 그룹에 active member로 join하지 않는다.
•
파티션 할당은 SplitEnumerator가 직접 수행 → Group Coordinator 입장에서 이 그룹은 멤버가 없는 상태
•
group.id가 하는 일은 두 가지
1.
체크포인트 시점 offset을 외부에 기록해 진행상황을 보여주는 가시성 라벨
2.
committed_offsets 시작 모드를 쓸 때의 시작 지점
즉 Kafka에서 Consumer Group = 병렬성과 리밸런싱의 주체라는 개념이 Flink에서는 통하지 않음
여러 해시 레이어
지금까지 본 것처럼 데이터는 파이프라인을 흐르며 서로 독립적인 여러 해시 기준으로 계속 재분배된다.
Flink to Redis 파이프라인을 예시로 보면,
Kafka Partition (입력 분할)
→ FLIP-27 split 할당 → source subtask
→ keyBy: murmurHash(key.hashCode()) % maxParallelism → Key-Group → subtask
→ (Redis Cluster 싱크) CRC16(key) % 16384 → 슬롯 → 샤드
JavaScript
복사
Redis Cluster 싱크 단계에서도 마찬가지다.
•
Flink가 어느 subtask에서 쓰든, Redis는 key의 CRC16 해시로 슬롯을 정해 샤드에 분배
•
샤드가 증설(리샤딩)돼도 클라이언트가 MOVED 리다이렉트로 토폴로지를 갱신 → 애플리케이션은 샤드 수를 알 필요 없음
•
hash tag({key})를 쓰면 특정 key들을 의도적으로 같은 슬롯에 모을 수도 있음
정리하면 Kafka를 읽는 과정은 하나의 분할 기준이 아니라 세 개의 독립적인 층위로 이뤄진다.
•
Partition: 입력을 어떻게 나눠 읽을지 (Kafka가 정하고 Flink는 FLIP-27로 직접 할당받음)
•
Key-Group: State를 어떻게 나눠 유지할지 (Flink가 정하고 rescale에도 안정적이도록 고정 버킷으로 관리)
•
Offset: 어디까지 처리했는지 (Kafka가 아니라 Flink 자신의 Checkpoint)
이 셋을 분리해서 보면,
•
파티션을 늘렸는데 왜 처리량이 그대로지인지
•
parallelism을 바꿨는데 왜 재시작이 오래 걸리는지
•
Kafka lag 대시보드랑 Flink 지표가 왜 다르게 보이는지
같은 질문에 답할 수 있다. 다음 글에서는 이 구조 위에서 Autoscaler가 실제로 무엇을 관측해 parallelism을 조정하는지 다룬다.