Problem RestatementProblem
Every time an ad is shown (an impression) or clicked, the app sends an event like { event_id, ad_id, campaign_id, user_id, type: click, timestamp }. Advertisers want dashboards showing clicks, impressions and click-through rate (CTR = clicks ÷ impressions) per ad and campaign, minute by minute. Advertisers are billed from these numbers, so counts must be correct: no double counting and no lost events.
RequirementsRequirements
1.1 Functional
- Ingest impression and click events.
- Aggregate counts per ad and campaign per minute, and serve queries for any time range.
- Filter out duplicates and obvious bots.
- Show near-real-time numbers on dashboards, and final exact numbers for billing.
1.2 Non-Functional
- Accuracy: exact for billing, near-exact for live dashboards.
- Freshness: dashboards within about 1 minute of real time.
- Scale: billions of events per day.
- Durability: raw events kept so we can recompute when a bug is found.
1.3 Scale Estimates
- 10 billion impressions + 200 million clicks per day ≈ 120,000 events/sec, peaks of 500K/sec.
- Event size ≈ 200 bytes → about 2 TB/day raw.
- Aggregates: 10M active ads × 1,440 minutes → up to 14B rows per day at 1-minute detail. We roll these up to hourly and daily after a few days.
1.4 API Design
/v1/events(batched from ad servers and SDKs), mostly written straight to Kafka./v1/stats?campaign_id=77&from=...&to=...&granularity=hour→ [{ time, impressions, clicks, ctr }]High-Level ArchitectureArchitecture
2.1 Overview
- Collectors: receive events and write them to Kafka.
- Kafka: durable buffer, partitioned by
ad_id. - Raw storage: all events copied to object storage (S3) for replays and audits.
- Stream processor (Flink): deduplicates, counts per ad per minute, and writes aggregates.
- OLAP store (ClickHouse, Druid or Pinot): an analytics database built for fast "sum these numbers over a time range" queries.
- Batch job (Spark, hourly/daily): recounts from raw storage to produce the final billing numbers.
- Query service: serves dashboards.
2.2 Architecture Diagram
%%{init: {"look":"handDrawn","handDrawnSeed":7,"theme":"base","fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","themeVariables":{"fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","fontSize":"16px","primaryColor":"#fff4e6","primaryBorderColor":"#1e1e1e","primaryTextColor":"#1e1e1e","secondaryColor":"#e7f5ff","tertiaryColor":"#ebfbee","lineColor":"#1e1e1e","textColor":"#1e1e1e","mainBkg":"#fff4e6","nodeBorder":"#1e1e1e","clusterBkg":"#f8f9fa","edgeLabelBackground":"#ffffff","classText":"#1e1e1e"}}}%%
flowchart LR
AS["Ad servers / SDKs"] --> COL["Collectors"]
COL --> K[("Kafka - by ad_id")]
K --> FL["Stream processor - dedupe + 1-min counts"]
FL --> OL[("OLAP store")]
K --> S3[("Raw events - S3")]
S3 --> SP["Daily batch recount"]
SP --> OL
SP --> BILL[("Billing tables")]
DASH["Advertiser dashboard"] --> QS["Query Service"]
QS --> OLData ModelData model
Raw event (Kafka / S3):
event_id, type (impression|click), ad_id, campaign_id, user_id, ts, ip, user_agent
Aggregate table (OLAP):
ad_id, campaign_id, minute, impressions, clicks, source (stream|batch)
primary sort: (campaign_id, ad_id, minute)Key FlowsFlows
- The ad server sends an event with a unique
event_id(created when the ad is served). - The collector writes it to Kafka and acknowledges. Kafka keeps it for 7 days.
- Flink reads events, drops duplicates by
event_id(keeping seen IDs for a few hours in its state), and adds to the counter for(ad_id, minute). - When a minute is complete (after the watermark, explained below), Flink writes that minute's row to the OLAP store.
- The nightly batch recounts the day from S3 and overwrites the stream numbers for that day. Billing uses only batch numbers.
Deep Dive A — Counting each event exactly onceDeep dive
Advertisers are billed from these counts, so a duplicate is an overcharge and a dropped event is lost revenue. Events arrive more than once routinely: the SDK retries, the collector retries, a processor restarts.
Increment a row per event
Every event does UPDATE stats SET clicks = clicks + 1 WHERE ad_id = ? AND minute = ?.
%%{init: {"look":"handDrawn","handDrawnSeed":7,"theme":"base","fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","themeVariables":{"fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","fontSize":"16px","primaryColor":"#fff4e6","primaryBorderColor":"#1e1e1e","primaryTextColor":"#1e1e1e","secondaryColor":"#e7f5ff","tertiaryColor":"#ebfbee","lineColor":"#1e1e1e","textColor":"#1e1e1e","mainBkg":"#fff4e6","nodeBorder":"#1e1e1e","clusterBkg":"#f8f9fa","edgeLabelBackground":"#ffffff","classText":"#1e1e1e"}}}%%
sequenceDiagram
participant S as Ad SDK
participant C as Collector
participant DB as Stats table
S->>C: click event e-991
C->>DB: clicks = clicks + 1
C--xS: ack lost
S->>C: retry click e-991
C->>DB: clicks = clicks + 1
Note over DB: one click billed twiceThe write carries no identity, so the database cannot tell a retry from a real second click. A processor restart replays a chunk of the Kafka log and inflates every counter in it. At 120,000 events/sec the row is also a contention point long before correctness matters.
Deduplicate by event ID, and checkpoint the processor
Give every event a unique event_id when the ad is served. The stream processor keeps the IDs it has seen for a few hours in its state and drops repeats. Flink checkpoints that state together with its Kafka offsets, so a restart resumes from a consistent point rather than re-reading blindly.
That removes both duplicate sources within the window. What it does not survive is anything outside the window: an event that arrives 12 hours late, a bug in the counting logic shipped last Tuesday, or a state backend that has to be rebuilt from scratch. The counters are now correct-ish and unfixable.
Idempotent writes, plus a batch recount that owns the billing numbers
Two changes on top of the above:
- Write the row, do not increment it. The processor emits "
(ad_id, minute)= 4,182", not "+1". Replaying the same minute writes the same value, so a retry at the storage layer costs nothing. Idempotency moves from "we deduplicated upstream" to "the write itself is safe to repeat". - Keep every raw event in object storage, and recount nightly. A Spark job reads the day's raw events, counts them exactly, and overwrites the streamed rows. Dashboards read the stream numbers for freshness; billing reads only the batch numbers.
%%{init: {"look":"handDrawn","handDrawnSeed":7,"theme":"base","fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","themeVariables":{"fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","fontSize":"16px","primaryColor":"#fff4e6","primaryBorderColor":"#1e1e1e","primaryTextColor":"#1e1e1e","secondaryColor":"#e7f5ff","tertiaryColor":"#ebfbee","lineColor":"#1e1e1e","textColor":"#1e1e1e","mainBkg":"#fff4e6","nodeBorder":"#1e1e1e","clusterBkg":"#f8f9fa","edgeLabelBackground":"#ffffff","classText":"#1e1e1e"}}}%%
flowchart LR
K[("Kafka")] --> FL["Stream - dedupe, 1-min counts"]
K --> S3[("Raw events - object storage")]
FL -->|"set minute = N"| OL[("OLAP store - dashboards")]
S3 --> SP["Nightly recount"]
SP -->|"overwrite the day"| OL
SP --> BILL[("Billing tables - source of truth")]This is what makes a counting bug survivable. Fix the code, re-run the job over the raw events for the affected week, and the numbers correct themselves — which is the real reason the raw copy is worth its storage cost.
Deep Dive B — Late events and hot adsDeep dive
- Late events: a phone that was offline can send a click 10 minutes late. Flink uses a watermark ("we assume all events up to time T have arrived"), for example 2 minutes behind real time. It then allows a longer "allowed lateness" window where it updates already-written rows. Events later than that are still in S3, so the nightly batch counts them.
- Hot ads: a Super Bowl ad can get 100K events/sec on one
ad_id, which overloads one partition. Pre-aggregate: each collector sums locally for 1 second and sends{ad_id, +clicks, +impressions}. Or add a random suffix to the key for hot ads and combine the pieces later. - Click fraud: filter bots before counting (known bad IPs, too many clicks per user per minute). Keep the filtered events so the rules can be audited.
Trade-offs & AlternativesTrade-offs
| Decision | Choice | Why | Alternative |
|---|---|---|---|
| Architecture | Stream for speed + batch for truth (Lambda) | Fresh dashboards and exact billing | Stream only (Kappa): one codebase, harder to guarantee exactness |
| Store | OLAP (ClickHouse/Druid) | Fast time-range sums | SQL DB: slow at billions of rows |
| Dedup | By event_id in stream state | Simple and exact within the window | Probabilistic (Bloom filter): less memory, rare mistakes |
| Hot keys | Pre-aggregate at collectors | Removes skew early | Key salting: more merge work |
Common Follow-up QuestionsFollow-ups
- "How do you fix a bug that miscounted last week?" Fix the code and re-run the batch job on the raw events in S3 for that week. This is why we keep raw data.
- "How do you count unique users who clicked?" Use HyperLogLog sketches per ad per hour. They merge easily and use tiny memory, with about 1% error.
- "Query latency?" Pre-aggregate to hour and day tables, and sort data by campaign so one campaign's rows are stored together.
Wrap-UpWrap-up
Send events with unique IDs into Kafka and keep a raw copy in S3. A stream processor deduplicates and counts per ad per minute with checkpoints, and writes idempotently to an OLAP store for dashboards. A nightly batch recount from raw data produces the exact billing numbers. Handle late events with watermarks plus the batch path, and pre-aggregate hot ads.