Data Eng Wiki

From Homegrown to Flink - Migrating a Stateful Ad Event Join

13 min read

From Homegrown to Flink: Migrating a Stateful Ad Event Join at Scale

Author: Aleksandr Birin (Zalando Marketing Services) | Source: engineering.zalando.com | Published: 2026-07-24

한 줄 요약

Flink의 고수준 interval join이 요소마다 RocksDB seek를 발생시켜 CPU의 15~30%를 먹는 것을 flame graph로 발견하고, 저수준 KeyedCoProcessFunction으로 직접 매칭 로직을 써서 리소스를 1/3로 줄였다.

핵심 주장/내용

  • 문제 도메인: 고객이 sponsored 아이템을 보거나 클릭하면 그 상호작용을 그것을 배치한 원래 auction과 매칭해야 한다. auction 레코드는 ad server의 delivery 결정 컨텍스트를, 상호작용 이벤트는 결과를 담고 있고, 둘을 조인하는 것이 원시 클릭을 청구 가능한 요금으로 바꾸고 ad serving 시스템으로의 피드백 루프를 닫는다. 근실시간이어야 하고 15분 윈도우로 제한 — 그 후 도착하는 상호작용은 버려진다. 이벤트를 놓치면 광고주 액션이 미청구되므로 match rate가 비즈니스 크리티컬 지표
  • 7년 전 homegrown 솔루션의 caveat 2가지: Kinesis와 Nakadi(Kafka 위 추상) 스트림을 소비해 bid 이벤트를 기본 인메모리 캐시에 저장하고, 상호작용 이벤트마다 키로 대응 bid를 찾아 매칭·enrich·billing 이벤트 생산. 최대 200MB/s 부하라 여러 pod 사용
    1. pod를 두 스트림에 걸쳐 깔끔히 파티셔닝할 수 없어 각 pod가 자기 몫의 bid 이벤트에 더해 모든 상호작용 이벤트를 읽어야 했다
    2. 메모리 상태 백업이나 체크포인팅이 없어 매 배포·rescale 연산마다 저장된 이벤트를 잃었다
    • → 이것이 rescaling을 피하려는 pod 과프로비저닝으로 이어졌다
  • 대안 평가: 옛 앱 개선(체크포인팅과 정렬된 스트림 파티셔닝 구현 가능하지만 더 나은 도구가 이미 존재), Nakadi SQL(PoC에서 이벤트 match rate가 ~50%에 그쳐 비현실적. 명확한 설명 없이 상당 부분을 버려서 진단에 더 투자하지 않음), Spark(Zalando에 채택되어 있지만 둘 다 평가한 다른 팀들이 Flink의 레이턴시와 오토스케일링 동작이 더 낫다고 보고해 배제), Flink(이미 여러 팀에서 프로덕션. keyed state, event-time join, 체크포인팅, 수평 파티셔닝이 유스케이스에 잘 맞음. 단점은 Tech Radar TRIAL 상태와 팀 경험 부족 — Zalando Flink guild와 다른 팀의 프로덕션 배포가 리스크를 수용 가능하게 만들었다)
  • 두 가지 개선: RocksDB state backend + 3분 증분 체크포인트 — 재시작·rescale 시 최대 15분치 버퍼 이벤트를 잃는 대신 마지막 체크포인트에서 복구. 이것이 안전한 오토스케일링을 가능하게 한 것. 그리고 bid만이 아니라 양쪽 스트림의 상태를 버퍼링 — homegrown은 대응 bid보다 먼저 도착한 상호작용 이벤트를 버렸는데, 양쪽을 버퍼링하면 이 out-of-order 케이스를 처리해 match rate가 0.5% 증가
  • 인프라 — AWS 관리형 vs K8s: 빠른 프로토타이핑을 위해 관리형을 골랐다가 단순하지만 유연하거나 투명하지 않음을 빠르게 발견. 관리형은 오토스케일링 튜닝이 제한적(CPU만), 지표가 CloudWatch만, Zalando 생태계 통합 제한, Flink v1만 지원. K8s는 더 낮은 비용, 클러스터 설정·환경·오토스케일러 완전 통제, 좋은 인프라 통합, Flink v2 지원 → 나중에 K8s로 이전
  • Interval join의 함정 — flame graph가 결정적이었다: 초기에는 interval join이 잘 맞을 것 같았다(keyed 파티셔닝과 필요한 시간 경계를 지원). 시스템은 돌았지만 status quo 애플리케이션보다 많은 worker와 리소스를 썼다
    • Flink UI의 Flame Graph에서 결정적 발견 — interval join이 CPU의 약 15~30%를 seek 연산에만 쓰고 있었다. 구현을 보니 매칭 경로가 레코드를 읽기 전에 RocksDB iterator를 열고 seek를 수행하는데, 파티션에 항목이 몇 개든 상관없이 그렇다. bid ID가 고유하므로 각 파티션에 정확히 하나의 항목만 있는데도 처리되는 모든 요소마다 seek가 발생한다. seek는 디스크의 정렬된 파일에서 시작 위치를 찾아야 하므로 이 처리량에서 요소당 하나의 seek는 상당한 오버헤드
    • 두 번째 seek 원천: interval join은 인메모리 캐시가 있는 RocksDB 기반 정렬 집합에 만료 타이머를 저장한다. 이 처리량의 15분 윈도우에서 이 큐가 영구적으로 약 7,000만 항목까지 자라고, 캐시가 소진되면 Flink가 RocksDB로 seek해 다음 배치를 로드하면서 끊임없는 스캔의 두 번째 원천을 만든다
    • 캐시 블록 크기 증가나 인메모리 상태 같은 최적화 옵션이 있어도 이 접근은 알고리즘적으로 이 시나리오에 부적합하다
  • 왜 KeyedCoProcessFunction이 해결하나: bid 이벤트용과 상호작용 이벤트용 두 개의 ValueState를 만들고 둘 다 bid ID로 파티셔닝. bid ID가 고유하므로 각 조회가 iterator도 seek도 없는 RocksDB 직접 point lookup이다. 상태 만료는 명시적 타이머 대신 백그라운드의 RocksDB TTL compaction으로 처리되어 요소별 타이머 등록도, 유지할 7천만 항목 큐도 없다. watermark 전진이 타이머 큐를 스캔할 필요가 아예 없어져 두 번째 seek 원천도 제거
  • RocksDB 상태 증가 문제와 튜닝: interval join에서 low-level process join으로 옮긴 뒤 상태 증가와 캐시 정리 문제. 타이머가 없으니 다른 정리 방법이 필요했고, 증분 체크포인트를 쓰므로 out-of-the-box 정리 옵션은 state compaction 중뿐이었다. 기본 설정이 안 맞아 많은 튜닝이 필요했다
    • write buffer와 compaction 파일 크기를 늘려 compaction이 덜 자주 돌되 실행당 더 많은 데이터를 처리하게 해서 TTL 필터가 만료된 키를 수집할 기회를 더 주었다 (writebuffer.count: 16, writebuffer.size: 512MB, compaction.level.target-file-size-base: 256MB, compaction.level.max-size-level-base: 1GB)
    • rescale 후 RocksDB 파일에 더 이상 그 subtask에 속하지 않는 키가 남으므로 incremental-restore-async-compact-after-rescale와 rescaling.use-delete-files-in-range로 async compaction·range delete를 촉발해 공간을 즉시 회수
    • SSD 최적화 설정과 LZ4 압축(compression.per.level: LZ4_COMPRESSION, predefined-options: FLASH_SSD_OPTIMIZED)
  • 인프라 함정들: Kinesis Flink 커넥터에 Protobuf aggregated event 읽기 지원이 놀랍게도 없어서 Flink 안에 자체 deaggregation 레이어를 써야 했다. Enhanced Fan-Out을 켜기 전까지 간헐적 ReadProvisionedThroughputExceeded — shard가 충분했는데도 KCL v3 로드밸런싱 알고리즘이 부하를 제대로 분배하지 않았다(consumer 3개: status quo 앱, Firehose archiver, Flink). Nakadi 연결은 애플리케이션당 제한이 있고 parallelism으로 설정된 각 subtask가 자기 연결을 가지므로 한도에 쉽게 도달할 수 있다
    • pod eviction — Karpenter(K8s 노드 오토스케일러)가 rescaling 중 pod가 스트림 소비 준비를 마치기 전에 evict했다. Flink에 자기 오토스케일러가 있으므로 K8s가 간섭하면 안 된다. 적절한 annotation(또는 PodDisruptionBudget)으로 해결
    • OOM kill — Java OutOfMemoryError가 아니라 Linux 커널이 프로세스를 종료시키고 있었다. Flink pod가 단일 JVM 프로세스로 돌므로 커널 kill이 graceful handling 없이 전체 pod를 무너뜨렸다. pod 메모리를 16GB → 20GB로 올려 JVM 프로세스 크기 위에 4GB 버퍼를 만들고, auto-tuning이 이전에 권장했지만 옳지 않았던 jvm-overhead.fraction(0.063 → 0.2), jvm-overhead.max(1GB → 4GB), managed.fraction(0.83 → 0.6)을 조정. RocksDB가 native 메모리도 쓰므로 버퍼가 필요하다
  • 마이그레이션 — 4주 shadow + 1주 A/B: 전용 입출력 스트림을 가진 완전한 shadow 파이프라인을 만들고, 다운스트림 서비스에서 shadow 스트림도 읽어 메인 파이프라인과 병렬로 shadow 테이블에 쓰게 해 고객 영향 없이 실제 프로덕션 데이터를 비교. 검증 항목: 이벤트 enrichment 성공률(Flink 출력이 평균 99.9%로 status quo와 일치, 임계는 99~101%), 출력 정확성(캠페인별·일별로 집계해 프로덕션 테이블과 비교), 처리 레이턴시(auction 개시부터 Flink 출력까지, 15분 윈도우 내), 중복률(출력 스트림에서 event ID 기준 약 0.2%). shadow 지표가 안정된 뒤 1주 A/B로 캠페인 결과를 end-to-end 검증하고 config 업데이트로 전환

주요 수치 / 사실

  • join operator 리소스 (subtask당 5GB 메모리 + 1 CPU 코어): Before subtask 3060개, load 30%, 메모리 150300GB, CPU 30~60코어 → After subtask 최대 10개, load 15%, 메모리 50GB, CPU 10코어
  • 전체 애플리케이션 결과: 평균 pod 수 상시 20 → 평균 5(범위 1~8), CPU 30 → 약 20코어(Flink가 homegrown보다 상태 관리에 CPU를 더 쓴다), 메모리 320GB → 약 100GB, EC2 비용 일 약 €80 → €30
  • 이벤트 enrichment 성공률 +0.5% (two-way 매칭 알고리즘)
  • 가장 중요한 개선은 체크포인팅이었다 — 이전에는 재시작·rescale마다 최대 15분치 버퍼 이벤트를 잃었지만 3분 증분 체크포인트로 마지막 체크포인트에서 복구해 상태 손실 없이 공격적 오토스케일링을 안전하게 만들고 pod 과프로비저닝 필요를 없앴다
  • 저수준 접근이 매칭·만료 알고리즘에 대한 완전한 통제를 주었고 그것이 이 효율 이득의 열쇠

관련 위키

Source: 원문 보기

그래프 뷰

백링크