Skip to content

ZALANDO

Read original ↗

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

Summary

Zalando's Ad Platform team (ZMS — Zalando Marketing Services) migrated a 7-year-old homegrown stateful event-join application to Apache Flink. The system matches ad auction bid events with user interaction events (views/clicks) within a 15-minute window to produce billable events at up to 200 MB/s throughput. The migration went through multiple architectural iterations — from AWS Managed Flink to self-hosted K8s, from Flink's built-in Interval Join to a custom KeyedCoProcessFunction — and required deep RocksDB compaction tuning, Kinesis connector workarounds, K8s pod eviction fixes, and JVM memory layout tuning. The result: 75% fewer pods (20 constant → 5 average), 69% memory reduction (320 GB → 100 GB), 62% cost reduction (€80 → €30/day), and a +0.5% event match rate improvement from two-way state buffering.

Key Takeaways

  1. Interval Join is not suited for unique-key joins at high throughput. Flink's Interval Join opens a RocksDB iterator and performs a seek on every processed element — even when each partition holds exactly one entry. At Zalando's throughput, this consumed 15–30% of CPU on seek operations alone. Additionally, ~70 million expiration timers in a sorted set generated a second constant-scan source (Source: flame graph analysis).

  2. KeyedCoProcessFunction with point lookups eliminates seek overhead. By implementing custom join logic with two ValueState objects (one per stream) partitioned by bid ID, every lookup becomes a direct RocksDB point-get — no iterator, no seek. State expiry via RocksDB TTL compaction replaces explicit timer registration entirely (Source: flame graph comparison showing seeks eliminated).

  3. RocksDB TTL compaction requires non-default tuning for streaming workloads. Default RocksDB settings caused unbounded state growth because compaction ran too infrequently to exercise the TTL filter. The fix: 16 write buffers (default 2), 512 MB write buffer size (default 64 MB), 256 MB target file size base (default 64 MB), LZ4 compression, and SSD-optimized presets (Source: RocksDB configuration table).

  4. Two-way state buffering increases match rate. The homegrown solution only buffered bid events, dropping interactions that arrived before their corresponding bid. Buffering both sides handles out-of-order delivery and increased match rate by 0.5% — directly impacting billing revenue (Source: match rate graph).

  5. Incremental checkpoints enable safe autoscaling. 3-minute incremental checkpoints to RocksDB state backend mean any restart recovers from last checkpoint instead of losing up to 15 minutes of buffered events. This made aggressive autoscaling safe (range 1–8 pods vs constant 20) without state loss (Source: results section).

  6. Shadow pipeline validation over 4 weeks before cutover. Full shadow pipeline with dedicated I/O streams, compared against production at 99.9% enrichment success rate, followed by a 1-week A/B test confirming campaign outcome equivalence (Source: migration section).

  7. Kinesis aggregated-record deaggregation required custom implementation. The Flink Kinesis connector library lacked Protobuf-based deaggregation support for AWS KPL-aggregated records — Zalando built their own layer inside Flink (Source: Kinesis connector section).

  8. K8s autoscaler interferes with Flink's own autoscaler. Karpenter (K8s node autoscaler) evicted Flink pods during Flink-initiated rescale operations, causing stream consumption failures. Fix: add annotation blocking K8s interference, or use a PodDisruptionBudget (Source: pod evictions section).

  9. JVM memory overhead needs explicit buffering for RocksDB native memory. Flink auto-tuning recommendations for memory parameters were insufficient — kernel OOM kills (not Java OOMError) terminated pods. Fix: increase pod memory from 16 GB to 20 GB, set jvm-overhead.max to 4 GB (default 1 GB), reduce managed memory fraction from 0.83 to 0.6 to leave room for RocksDB's native allocations (Source: memory tuning table).

  10. Enhanced Fan-Out required for multiple Kinesis consumers. Standard Kinesis throughput was insufficient for 3 consumers (app + Firehose + Flink shadow) despite adequate shard count — ReadProvisionedThroughputExceeded errors persisted until Enhanced Fan-Out was enabled (Source: Kinesis section).

Operational Numbers

Metric Before (homegrown) After (Flink)
Pod count 20 constant 5 average (range 1–8)
Memory 320 GB ~100 GB
CPU cores — ~20 (higher per-pod due to state management)
EC2 cost ~€80/day ~€30/day
Event match rate baseline +0.5%
Checkpoint interval none (state lost on restart) 3 min incremental
State loss on restart up to 15 min last checkpoint only
Throughput up to 200 MB/s up to 200 MB/s

Join operator resource comparison (KeyedCoProcessFunction vs Interval Join):

Metric Interval Join KeyedCoProcessFunction
Subtasks 30–60 10 (max)
Load 30% 15%
Memory 150–300 GB 50 GB
CPU cores 30–60 10

Systems and Concepts Extracted

Systems

Concepts

Patterns

Caveats

  • The article focuses on the near-real-time stream only; correct billing and campaign reporting also rely on a separate batch pipeline (lambda architecture — not detailed here).
  • Resource numbers for the join operator only (5 GB memory, 1 CPU per subtask); full application includes source operators, deduplication, and sink.
  • Kinesis connector limitations were for Flink v1 only; migration to Nakadi + Flink v2 resolved these.
  • The Nakadi SQL PoC achieved only ~50% match rate without clear diagnosis — not explored further.

Source

Last updated · 595 distilled / 1,807 read