Problem
Design a system that aggregates ad click events for an ad network in near-real-time, at a scale of roughly 10 billion events per day, to power both billing and real-time campaign metrics.
Requirements
Functional
- Ingest click events at up to ~10B/day and aggregate counts per (ad_id, time window) with low latency, not batch-hours-later latency.
- Serve both a near-real-time view (last few minutes) and historical rollups (hourly/daily).
Non-functional
- No double-counting on retries or redelivery — billing-relevant aggregates need exactly-once semantics in effect, even if not literally at the transport layer.
- Tolerate late-arriving events (client batching, mobile network delay) without silently misattributing them to the wrong time bucket.
- Handle "hot" ad_ids (a viral campaign) without a single partition or key becoming a bottleneck.
- Durable storage for audit/reprocessing, queryable at multiple granularities.
Architecture
- Ingestion: a load-balanced API validates and stamps each click, then publishes to a partitioned log (Kafka/Pub-Sub) keyed by ad_id.
- Stream aggregation: a windowed stream job (Flink/Kafka Streams) groups by (ad_id, time bucket) using event-time watermarks to handle late data correctly.
- Deduplication: a client-generated idempotency key is checked against a bounded-window dedup store (Redis/RocksCE state store) before a click reaches the aggregator.
- Storage: aggregates flush periodically to a columnar warehouse for historical queries; a fast KV cache holds the last few minutes for near-real-time reads.
- Hot-key handling: a local combiner pre-aggregates at the ingestion node before the keyed shuffle, so a viral ad's clicks are pre-summed per node instead of funneling every click through one downstream partition.
Key decisions
- At-least-once delivery with idempotent dedup, not end-to-end exactly-once — true exactly-once across a distributed pipeline is expensive and often illusory; deterministic dedup downstream is cheaper and more robust.
- Event-time windowing with watermarks over processing-time windowing, so a click arriving 30 seconds late still lands in the correct bucket instead of being misattributed to whenever it happened to arrive.
- Static, uniform edge pre-aggregation over dynamic runtime repartitioning — simpler to reason about, at the cost of some memory per ingestion node.
Tradeoffs
- The watermark lateness bound (e.g. 2 minutes) is a real dial: too short and you undercount, too long and you delay when a bucket is considered final.
- Local pre-aggregation reduces hot-key pressure but adds a small buffering latency and per-node combiner-state complexity.
- The dedup window is bounded (e.g. last 10 minutes of event IDs) — a duplicate arriving after that window is not caught, an accepted risk rather than an unbounded guarantee.
Failure modes
- Ingestion node crash mid-batch loses that node's un-flushed partial aggregate — mitigated by frequent (sub-second) flushing or checkpointed combiner state.
- Stream processor restart relies on the platform's checkpointing (offsets + operator state) to resume without reprocessing the whole stream.
- Downstream storage outage: the log buffers consumption up to its retention window; an outage longer than retention loses data, so retention has to be sized for a realistic outage, not just steady-state lag.
Capacity estimates
- 10B events/day ≈ ~115K events/sec average; realistic peak-to-average ratios for ad traffic put peak around 3–5x average, so provisioning for roughly 350K–500K events/sec at peak.
- Each event is a few hundred bytes, so raw volume is on the order of a few TB/day before compression.
- Aggregates are orders of magnitude smaller than raw events, so the hot-path KV store holds a small, bounded working set — ingestion and processing capacity is what the raw stream volume actually drives.
What I'd change at 10x scale
- At ~100B events/day, a single global aggregation tier stops being viable — shard by region with local aggregation and ship only rolled-up regional aggregates cross-region, not raw events.
- Move from a fixed lateness-bound watermark to a tiered reconciliation: a fast near-real-time tier for low latency, plus a slower batch pass hours later that recomputes exact aggregates from the durable raw log.
- Make hot-key pre-aggregation adaptive (detect a hot ad_id at runtime) rather than uniform, since applying it everywhere at 10x scale wastes resources on the long tail of ads that never get hot.