Design an ad click aggregation system

Count ad clicks per ad and per time window from billions of events a day, accurately enough to bill advertisers, with results available within minutes.

We are building the pipeline that turns raw ad clicks into per-ad, per-minute click counts that advertisers are billed on and that dashboards show within a few minutes. A click only counts if it can be tied back to the ad impression that was actually served.

Functional requirements

• Ingest click events from web and app clients via a click redirect service, and impression (ad served) events from the ad server.
• Join each click to its impression by impression_id; a click with no matching impression within the attribution window (e.g. 1 hour) is not billable.
• Count clicks per ad per 1-minute bucket, rolled up to hour and day, sliced by a small fixed set of dimensions (campaign, advertiser, country, device type).
• Advertiser dashboard: "clicks for ad X (or campaign Y) between T1 and T2, by minute/hour/day".
• Correct counts when events arrive late, are duplicated, or must be replayed after a bug.
• Hook for invalid-traffic filtering: clicks flagged as bots are counted separately, not billed.

Non-functional requirements

• Billing accuracy: each valid click counted exactly once in the final numbers. Dashboards may be a little behind but never double-count.
• Freshness: p95 under 2 minutes from click to dashboard; final (billing-grade) numbers within a few hours.
• Durability: no click lost once acknowledged; raw events retained so any window can be recomputed.
• Scale: billions of events a day, skewed (one viral ad can take 1% of all clicks).
• Reads are cheap and fast: dashboard queries under 200 ms p99.

Back-of-the-envelope estimation

• 10B impressions/day, 1B clicks/day (10% of the impression volume is a generous CTR to size for). Average 12K clicks/s, peak 5× ≈ 60K clicks/s; impressions ≈ 115K/s average, 600K/s peak.
• Event size ≈ 300 B. Clicks: 1B × 300 B = 300 GB/day raw. Impressions: 3 TB/day. Kafka at 3-day retention with replication factor 3 ≈ 30 TB.
• Join state: impressions kept for a 1-hour window at peak ≈ 600K/s × 3,600 × ~100 B (id + ad + dims only) ≈ 216 GB, spread across the join workers' local RocksDB stores.
• Output: 10M active ads, but only ~2M get a click in a given hour. Per-minute rows ≈ 2M ads × 60 × few dimension combos ≈ 300M rows/hour before rollup; minute rows kept 7 days, hourly kept 2 years.

Producers publish to Kafka, not to an HTTP API, so events survive any downstream outage.

topic impressions  key=impression_id
{impression_id, ad_id, campaign_id, advertiser_id, user_id_hash, country, device, served_at}

topic clicks       key=impression_id
{click_id, impression_id, ad_id, user_id_hash, ip_hash, clicked_at, received_at}

The click redirect service (GET /c?imp=...&ad=...&sig=...) verifies the HMAC signature, writes the click to Kafka (acks=all), then 302-redirects to the landing page. It assigns click_id deterministically as hash(impression_id, clicked_at second) so a client retry produces the same id.

Dashboard read API:

GET /v1/ads/{ad_id}/clicks?from=2026-10-12T00:00Z&to=2026-10-12T06:00Z&granularity=minute|hour|day&group_by=country
200 {
  "ad_id": "ad_123",
  "series": [{"bucket": "2026-10-12T00:00Z", "country": "US", "clicks": 1840, "invalid": 12}],
  "watermark": "2026-10-12T05:58Z",   // buckets before this are complete for streaming
  "final_through": "2026-10-12T03:00Z" // buckets before this are billing-final
}
GET /v1/campaigns/{campaign_id}/clicks?...   same shape, summed across ads

Reads are idempotent GETs; a request outside retention returns 400 range_out_of_retention. The response always carries watermark and final_through, so the dashboard can grey out partial buckets instead of showing a dip.

Operator controls (internal):

POST /v1/admin/backfill {from, to, reason, source: "raw_archive"} -> {job_id}
GET  /v1/admin/pipeline/health -> {consumer_lag_s, watermark_lag_s, late_events_last_hour, join_miss_rate}

Why Kafka rather than a plain queue: we need ordered, partitioned, retained logs that several consumers (stream job, raw archiver, fraud scoring) read independently, and that we can rewind by offset to replay. A delete-on-ack queue gives none of that.

• Kafka topics impressions, clicks, attributed_clicks: the durable source of truth for the last 3 days, partitioned by impression_id so a click and its impression land in the same partition number (co-partitioned join).
• Raw archive (S3 / object store, Parquet) partitioned by event_date/hour: every event forever (or per legal retention). This is what backfill and reconciliation read.
• Join state (Flink keyed state on RocksDB): impression_id -> {ad_id, campaign_id, advertiser_id, country, device, served_at}, TTL 1 hour + allowed lateness. Plus a dedupe set click_id -> seen, TTL 24 hours.
• Aggregates (OLAP store: ClickHouse / Druid / Pinot):

Rows are written with a replacing/versioned engine (ClickHouse ReplacingMergeTree(version) keyed on the full key), so re-emitting a bucket overwrites instead of adding. An OLAP store is chosen because every query is "filter on ad or campaign, range on time, sum"; columnar storage sorted by (ad_id, bucket_start) answers that from a few granules.

• Ad metadata (MySQL/Postgres): ads(ad_id PK, campaign_id, advertiser_id, created_at), used for dimension lookups and auth on the read API.

flowchart LR
  AS[Ad server] --> KI[Kafka impressions]
  U[User clicks ad] --> CR[Click redirect service]
  CR --> KC[Kafka clicks]
  KI --> J[Flink join and dedupe]
  KC --> J
  J --> KA[Kafka attributed clicks]
  KA --> AG[Flink windowed aggregator]
  AG --> OL[(OLAP store)]
  KI --> AR[(Raw archive S3)]
  KC --> AR
  AR --> B[Batch recompute job]
  B --> OL
  B --> BL[(Billing tables)]
  OL --> Q[Query service]
  Q --> D[Advertiser dashboard]
  KC --> F[Invalid traffic scorer]
  F --> J

1. Serve: the ad server picks an ad, mints impression_id, and publishes an impression event. The click URL embeds a signed impression_id.
2. Click: the redirect service validates the signature, publishes the click keyed by impression_id, and redirects. Publishing is on the request path, so Kafka is sized for peak with acks=all and a small local disk buffer if the broker is slow.
3. Join before counting: a Flink job reads both topics keyed by impression_id. It stores impressions in keyed state; when a click arrives it looks up the impression, drops it if click_id was already seen, attaches the ad's dimensions, and emits an attributed click. A click that beats its impression waits in state up to 10 minutes.
4. Aggregate: a second Flink job keys attributed clicks by (ad_id, salt) and counts them in 1-minute event-time tumbling windows. It emits when the watermark passes the window end, then emits corrected rows for late events until allowed lateness expires.
5. Serve reads: the query service reads minute rows for ranges under 2 days, hourly rows beyond that, and caches closed buckets (they rarely change) for 60 s.
6. Reconcile: an hourly batch job recomputes every bucket from the raw archive with the same join and dedupe logic. Its output is the billing-final number and overwrites the streaming rows for that hour.

Stream join correctness and late arrivals. Event time, not processing time, decides the bucket: a click at 10:00:59 that arrives at 10:03 counts in the 10:00 minute. Watermark = max seen clicked_at minus 30 s (covers p99 network delay). Window results fire at the watermark (fresh dashboards) and again for each late event up to 1 hour of allowed lateness, each time re-emitting the whole bucket with a higher version, which the replacing table keeps. Events later than 1 hour go to a late_clicks side topic, are counted in a metric, and are picked up by the hourly batch reconcile, so billing still includes them. Joins: the impression state TTL is the attribution window (1 hour). Clicks that arrive before their impression are buffered 10 minutes; if no impression shows up they go to an unattributed topic, not the count. Watch join_miss_rate: a jump means the impression producer is broken, not that users stopped clicking.

Replay-safe counts and explicit dedupe. Duplicates come from three places: client retries (same click_id, handled by the deterministic id), producer retries (Kafka idempotent producer), and consumer restarts (handled by Flink checkpoints + transactional sinks). The dedupe set in Flink state holds click_id for 24 hours: 1B ids/day × ~40 B ≈ 40 GB across workers, fine in RocksDB. A Bloom filter in front would cut memory by ~10× but gives false positives that silently drop real billable clicks, so we keep an exact set. Sinks must be idempotent: aggregates are written as full bucket values with a version, never as +1 increments, so a replay that rewrites the same bucket is harmless. Billing tables are written by run, each row carrying run_id; a rerun replaces the partition atomically.

We use the last three together: streaming for freshness, batch for the billing number.

Hot-ad skew and partitioned rollups. A Super Bowl ad can take 50K clicks/s, more than one Flink subtask (~20K events/s with state access) can handle. Joins are keyed by impression_id, which is spread evenly, so skew only hurts the aggregation stage. There, key by (ad_id, salt) with salt = hash(click_id) % N, where N = 1 for normal ads and up to 16 for ads on a hot list (refreshed every minute from click-rate metrics). Each salted shard pre-aggregates its minute bucket; a small second stage sums the N partials per (ad_id, minute) before writing. Because shards emit full partial counts (not deltas), the merge stays idempotent. Kafka partitioning of attributed_clicks uses the same salted key so one broker partition doesn't become the hot spot.

Backfill, dimension changes and cardinality. Every count must be recomputable from raw: the batch job (Spark over the Parquet archive) uses the same join/dedupe library as the Flink job, so the two paths can't drift. Backfill is the incident response: if a bug miscounted 10:00–14:00, fix the code, run POST /admin/backfill for that range, and it rewrites those partitions with a new run_id; the dashboard shows the new numbers once the swap commits. Dimensions are a closed list reviewed like a schema change. Each new dimension multiplies row count: country (200) × device (5) is 1,000 combos per ad-minute at worst, but adding campaign changes nothing (it is a function of ad_id, so it lives in metadata and in its own rollup table), while adding something like placement_id or raw user agent would explode it. Rule: dimensions that are 1:1 with ad go to lookup tables; new free-form dimensions only get hourly rollups, behind a cap on distinct values (top 100 plus other).

Invalid traffic without building a fraud platform. A scorer reads the clicks topic and emits {click_id, verdict} for simple rules: same ip_hash clicking the same ad more than 5 times in 10 minutes, known data-center IP ranges, click-to-impression gap under 200 ms. The join job holds each click briefly (it is already waiting on the watermark) and tags it invalid if a verdict arrives. Invalid clicks are counted in their own column, so a later, better fraud model can reclassify them in the batch job without touching the pipeline.

• Lambda (stream + batch) over stream-only: two code paths to run, and the dashboard number can change once the batch result lands. In return billing comes from a recompute over immutable raw data, which survives any streaming bug. A shared library keeps logic identical across both.
• Join before counting: needs ~200 GB of impression state and adds up to 10 minutes of wait for out-of-order clicks; counting raw clicks would be simpler but would bill clicks for ads that were never served or were spoofed.
• Exact dedupe set over a Bloom filter: about 10× more memory, but no billable click is ever dropped by a false positive.
• 1-hour allowed lateness: bounded state and early finality versus a small tail of late clicks that only the batch job counts.
• Fixed dimension list: advertisers can't slice by arbitrary attributes in real time; in return row counts and query cost stay predictable.
• Overwrite-by-version sink: requires an OLAP engine with replace semantics and background merges (reads may briefly see two versions unless they use FINAL/argMax); increments would be simpler but are not replay-safe.
• Per-region pipelines (when expanding): each region aggregates its own clicks and the global view sums them at query time; cross-region joins are avoided by keeping impression and click in the region that served the ad, at the cost of a little lost attribution when users roam mid-session.

Mid-level (L4)

• Puts a durable log (Kafka) between clients and counting, and counts per ad per minute in a stream processor.
• Stores pre-aggregated rows in a store suited to time-range queries rather than counting raw events at read time.
• Knows duplicates and late events exist and proposes a click id for dedupe.

Senior (L5)

• Uses event time with watermarks and allowed lateness, and explains what happens to an event after the cutoff.
• Makes sinks idempotent (overwrite buckets, not increments) and keeps raw events for batch reconciliation that produces the billing number.
• Joins clicks to impressions before counting, handles click-before-impression ordering, and fixes hot ads with salted keys and a two-stage rollup.

Staff (L6+)

• Frames replay and backfill as the main recovery tool, with one shared join/dedupe library for stream and batch so they can't diverge.
• Exposes freshness (watermark, final_through) and pipeline health in the API so consumers know which numbers are final.
• Governs dimensions and cardinality, leaves a clean hook for invalid-traffic scoring, and has a plan for multi-region expansion.