Primary concepts and the hard part
Requirements
Events: 1M/sec → 86B/day Event size: ~200B → 17 TB/day raw Cardinality: 1M ads × 1,440 minutes = 1.4B aggregate rows/day (before dimensions) Query load: dashboards ~1,000/sec; reports lower volume, heavier
API / Model
High-level architecture
Every event enters through one ingest path and then splits in two: a hot path in Flink produces approximate numbers within seconds, and a cold path recomputes exact numbers from the raw archive overnight. Both write to the same OLAP store.
- Ad servers fire events at the regional ingest gateway without waiting for a reply. The gateway validates them, adds geo and device, stamps a receive time next to the event time, and writes to Kafka
ad.events, partitioned byad_id. Flink consumes the topic and dedupes onevent_idfirst. - Event-time windowing puts each event in a tumbling 1-minute bucket by when it happened, not when it arrived.
- The watermark,
max_event_time − δ, decides when a bucket closes. Flink emits the bucket once the watermark passes its end, emits an update for a late event still within the grace period, and sends anything later to a side output. - Aggregation adds up counts and builds HyperLogLog sketches for unique counts, then writes the fast, approximate rows to the OLAP store.
- The query service reads the OLAP store for dashboards, reports and the billing export, and caches recent windows in Redis.
The cold path reads the same topic. Raw events are archived as Parquet in object storage, partitioned by date and hour, and outlive Kafka's 7-day retention. Each night, reconciliation recomputes yesterday from the archive and overwrites the stream estimates in the OLAP store with the exact figures billing uses. In parallel, the fraud filter reads ad.events and flags suspicious clicks for that nightly run instead of deleting them.
Same pipeline at 10x. 10M events/sec is in the range of the largest ad platforms, so 10x is the realistic next tier. Dashed outlines mark what's new or reshaped compared with today's design.
| Today | At 10x | |
|---|---|---|
| Events | 1M/sec | 10M/sec |
| Raw volume | ~17 TB/day | ~170 TB/day |
| Aggregate rows | ~1.4B/day | ~14B/day |
| Dedupe keys for a 1-hour late window | ~3.6B | ~36B |
| Dashboard queries | ~1,000/sec | ~10,000/sec |
What changes, and the number that forces it
- Exact dedupe leaves the stream. Keyed state on every
event_idacross an hour-long late window is ~36B keys at 10x. Checkpoints get so large that recovering from a failure takes longer than the failure lasted. The stream keeps a Bloom filter per time bucket instead: cheap, approximate, and a rare false positive only drops a real click from the live number. Exact dedupe moves to the batch recompute, which billing already treats as authoritative. - Aggregation starts before the shuffle. Salting hot keys works for a few known viral ads; at 10x there are always some you didn't predict. Each task sums counts locally per
(ad, minute, dimensions)before the network shuffle, so a viral ad sends a handful of partial aggregates to its partition instead of millions of raw events. - The pipeline runs per region. Shipping ~170 TB/day of raw events to one place costs a fortune in cross-region bandwidth and buys nothing. Each region runs its own Kafka, stream job and raw archive, and ships only aggregates. HyperLogLog sketches are mergeable, so global unique counts still work.
- The batch recompute goes hourly. A nightly recompute over ~170 TB of raw events no longer fits in a night. Recomputing each hour from the archive, once that hour's late window has closed, delivers billing corrections within hours and keeps any failed run small.
- The OLAP store is tiered. 14B+ aggregate rows a day on local SSD forever isn't affordable. Recent days stay on SSD; older segments live in deep object storage and load when queried, and minute → hour → day rollups kick in sooner.
- Dashboards read materialized summaries. At ~10k queries/sec, most dashboard loads ask the same per-advertiser questions. Those summaries are materialized as windows close and served from a result cache, leaving the OLAP store for reports.
What stays the same
Windows by event time, not processing time. Watermarks, the three-tier late-data policy, idempotent upserts keyed by (ad, window, dimensions), the same code for streaming and replay, an immutable raw archive, ingest that never blocks ad serving, and fraud that flags rather than deletes. The live number is still an estimate and the batch number is still the bill; only the boundary between them moved from nightly to hourly.
Trade-offs and deep dives
A click HAPPENS at 10:00:30 on a phone that's in a tunnel.
It ARRIVES at your ingest at 10:07:15.
Processing-time windowing → counted in the 10:07 bucket. WRONG.
Event-time windowing → counted in the 10:00 bucket. RIGHT.Advertisers are billed per minute and compare your numbers against their own. Attributing a click to the wrong minute is a billing dispute. Windowing must use the event's own timestamp.
But event-time windowing raises the question: when do you decide the 10:00 window is finished? You can't wait forever. That's what watermarks are for.
A watermark is the processor's assertion: "I believe all events with timestamp earlier than T have now arrived." It's typically computed as max_observed_event_time − allowed_lateness. When the watermark passes a window's end, the window fires and emits its result.
The trade-off is explicit and worth stating: a larger allowed-lateness δ captures more stragglers but delays every result by δ. A smaller δ gives fresher dashboards but drops or defers more late data. Pick δ from the observed distribution of arrival delay (e.g. δ = p99 of arrival_time − event_time), and say you'd measure it rather than guess.
Have an answer for each:
- Within the watermark — included normally.
- After the window fired but within a grace period — emit an updated result (a retraction plus a new value). Downstream stores must support upsert, which is why the OLAP layer is keyed by (ad, minute, dimensions) rather than append-only.
- Beyond grace — route to a side output and let the nightly batch job fix it. Don't distort the streaming pipeline to chase the long tail.
- Kappa (stream only, reprocess by replaying) is simpler and increasingly the default.
- Lambda (stream for speed, batch for truth) duplicates logic in two systems, which can drift.
For billing, the honest answer is a hybrid that leans Kappa: one stream pipeline produces the live numbers, and a reprocessing run of the same code over archived raw events produces the authoritative nightly figures. You get the correctness of a batch layer without maintaining two separate implementations. Stating it that way — same code, replayed — shows you understand why classic Lambda is criticized.
Flink's checkpointing gives exactly-once state semantics within the pipeline: on recovery, it restores state and rewinds Kafka offsets so no event is double-counted internally. But the moment you write to an external store, you need either a transactional sink or idempotent upserts keyed by (ad_id, window, dimensions). The end-to-end guarantee is effectively exactly-once, built from at-least-once delivery plus idempotent writes — the same framing as Module 5, applied to analytics.
Application-level dedupe on event_id is still required, because the ad server itself may retry and send the same event twice. That's outside Flink's guarantee entirely.
These queries scan billions of rows to compute sums grouped by a few dimensions. A row-oriented OLTP database reads entire rows off disk to sum one column. Columnar stores read only the columns referenced, compress each column separately (very effectively, since adjacent values are similar), and vectorize the scan. Orders of magnitude difference for exactly this access pattern. Never run these reports against the transactional database.
Storing every raw event forever in the query store is unaffordable. Pre-aggregate at ingestion to minute granularity, then roll minutes into hours after a few days and hours into days after a few months. Query granularity degrades with age, which matches how people actually use analytics — nobody needs minute-level data from eighteen months ago.
"Unique users who saw this ad" cannot be pre-aggregated by simple addition — you can't sum two unique-counts. Use HyperLogLog sketches, which are mergeable: the union of two sketches gives the unique count of the union, in a few KB, with ~2% error. If exact uniques are required for billing, compute them in the batch layer over raw data. This is a great place to show you know when approximation is acceptable and when it isn't.
The ad server fires events asynchronously and does not wait. If the analytics pipeline is down, ads still serve and events are dropped or buffered locally. Analytics completeness is subordinate to ad delivery — state that priority explicitly.
Gives per-ad ordering and lets stateful operators keep per-key state locally. The risk is a hot key: one viral ad concentrates on one partition. Mitigate by salting the key for known-hot ads (ad_123#0..9) and summing the sub-aggregates downstream — you trade a merge step for even distribution.