Design Ad Click Event Aggregation

Ad click aggregation counts how many times each ad was clicked in each time interval and makes those counts queryable. The counts drive billing (advertisers pay per click), so they must be accurate, and they drive dashboards and bidding decisions, so they must be fresh. The system is a data pipeline: high-volume ingest, stream processing, and an analytical store.

Requirements

Functional

• Aggregate the number of clicks per ad_id in the last M minutes (for example, 1-minute buckets).
• Return the top N most clicked ads in the last M minutes.
• Support filtering by attributes such as country or device type.
• Query historical aggregates (days to months back).

Non-functional

• Correctness. Counts are used for billing. Each click is counted exactly once: no duplicates, no losses. Late events must still be attributed to the right window.
• Freshness. Results within a few minutes of the click event.
• Query latency. Milliseconds for dashboards, at most a few seconds for filtered historical queries.
• Fault tolerance. Node failures must not change results. Data must be recoverable.
• Scalability. Handle peak load of several times the average, and skew where a few ads receive most clicks.

Estimation

Assumptions: 1 billion daily active users, 1 click per user per day on average, so 1 billion clicks per day.

• Average click rate: 1,000,000,000 / 86,400 ≈ 11,600 clicks per second (about 10,000).
• Peak: 5x average ≈ 50,000 clicks per second.
• Click event size (ad_id, click_timestamp, user_id, ip, country): about 0.1 KB. Daily raw volume: 1,000,000,000 × 0.1 KB = 100 GB per day, about 3 TB per month.
• Ingest bandwidth at peak: 50,000 × 0.1 KB = 5 MB per second. This is small; the challenge is correctness and aggregation, not bandwidth.
• Aggregated storage: assuming 2 million active ads and 1-minute buckets, 2,000,000 × 1,440 = 2.9 billion rows per day in the worst case, at roughly 30 bytes per row about 86 GB per day. Most ads are inactive in most minutes, so real volume is lower.

What the numbers imply

• 50,000 clicks per second at 0.1 KB is only 5 MB per second, so a handful of queue partitions could carry the bytes. Partition count is set by per-task processing (dedup lookups, state updates, roughly 5,000 to 10,000 events per second per task), so use about 32 to 64 partitions to leave headroom for skew and catch-up after an outage.
• Dedup state: at peak, 50,000 per second × 3,600 s TTL = 180 million click_ids at about 50 bytes each is roughly 9 GB, spread over partitions. That is too large for heap, so keyed state lives in RocksDB on local disk with incremental checkpoints.
• 2.9 billion aggregate rows per day is the worst case; raw storage is 3 TB per month, so raw events go to cheap object storage and only aggregates go to the serving store.

API design

Ingestion (click events reach the system from ad servers or a click-redirect service):

POST /v1/clicks
{ "click_id": "uuid", "ad_id": "ad123", "ts": 1780000000, "user_id": "u1", "ip": "...", "country": "US" }

In practice, ad servers write to a message queue directly rather than calling an HTTP endpoint per click.

Query APIs:

click_id is generated at click time (by the click-redirect service) and is used for deduplication.

Data model

Two kinds of data, with different storage needs.

Raw events. Kept for reconciliation, backfill, and reprocessing. They are written once, read in bulk, and never updated. Storage: object storage in a columnar format (Parquet), partitioned by hour. The message queue's retention (days) covers short-term replay.

Aggregated data. Small, read frequently, queried by ad and time range.

ad_click_agg(ad_id, click_minute, country, count)   -- primary key (ad_id, click_minute, country)
top_n_ads(window_size, update_time_minute, most_clicked_ads)

Storage choice reasoning:

A relational database can serve the aggregate table at moderate scale; a wide-column or OLAP store handles the write rate and time-range scans better.

High-level design

flowchart LR
    AS["Ad servers / click service"] --> Q1[("Queue: raw clicks")]
    Q1 --> AGG["Stream aggregation service"]
    Q1 --> RAW["Raw event sink"]
    RAW --> OBJ[("Object storage")]
    AGG --> Q2[("Queue: aggregated counts")]
    Q2 --> DB[("Aggregate database")]
    DB --> QS["Query service"]
    QS --> D["Dashboards / billing"]
    OBJ --> BATCH["Batch reconciliation"]
    BATCH --> DB

Component responsibilities:

• Queue for raw clicks (Kafka or similar, see Kafka internals). Decouples producers from processing, absorbs bursts, and allows replay. Partitioned by ad_id so one ad's clicks land on one partition.
• Stream aggregation service (Flink, Spark Structured Streaming, or a Kafka Streams application). Groups events into time windows and counts them.
• Queue for aggregated counts. A second queue between aggregation and the database lets the aggregation output be written exactly once using a transactional sink (see below).
• Aggregate database. Stores per-minute counts and top-N results.
• Query service. Reads from the aggregate store, rolls up minute buckets into larger ranges, and caches popular queries.
• Batch reconciliation. A periodic batch job recomputes counts from raw events in object storage and compares them with the streaming results.

Deep dive

1. Windowing and event time

Each click has an event time (when the click occurred) and a processing time (when the pipeline sees it). Billing needs event time, because a click at 10:59:59 that arrives at 11:00:03 belongs to the 10:59 bucket.

A tumbling window is a fixed, non-overlapping interval (one minute). A sliding window overlaps (last 5 minutes, updated every minute). Sliding windows for top-N are computed by combining tumbling one-minute counts: the count for the last M minutes is the sum of M one-minute buckets.

flowchart LR
    E["Click events"] --> M["Map: ad_id, minute bucket"]
    M --> C["Count per ad per minute (tumbling)"]
    C --> T["Top N per sliding window: merge sorted partial results"]

Top-N is computed in two steps: each partition computes its local top N for the window, and a final stage merges the local lists. Because partitioning is by ad_id, each ad's total is entirely in one partition, so merging the partial top-N lists is exact.

2. Late events and watermarks

Events can arrive late because of network delays, mobile clients that buffer, or retries. A watermark is the pipeline's estimate that no more events with event time earlier than some timestamp T will arrive. It is typically "max event time seen minus an allowed delay".

• A window closes and emits its result once the watermark passes its end.
• A longer allowed delay captures more late events but delays results.
• Events later than the watermark can be dropped, emitted as corrections, or sent to a side output.

The standard resolution for accuracy is to combine approaches: a short watermark (about a minute) for fresh results, and a nightly batch job that recomputes from raw events to correct any late data. This is the lambda architecture; a kappa architecture instead reprocesses by replaying the same stream through the same code. The trade-off is operating two code paths versus replaying large volumes through one.

3. Exactly-once counting

Counting twice bills advertisers wrongly; missing a click loses revenue. Failure points:

• Duplicate events at ingestion. The same click can be sent twice (client retry, redirect replay). Deduplicate on click_id: keep a set of recently seen ids in the stream processor's keyed state, with a TTL equal to the maximum expected duplicate delay. Because the queue is partitioned by ad_id, all copies of a click reach the same partition and the same state.
• Aggregator crash mid-window. After restart, some events could be re-read and counted again. Stream processors handle this with periodic checkpoints that store operator state together with the input offsets. On recovery, both are restored together, so events after the checkpoint are re-read and re-applied to the restored state only once.
• Output written twice. If the aggregator writes results to the database and then crashes before the checkpoint, the output is written again after restart. Solutions: (a) a transactional sink that commits output atomically with the checkpoint (two-phase commit, as in Flink's Kafka sink with transactions); or (b) idempotent writes, using (ad_id, minute) as the key so rewriting the same window overwrites rather than adds.

Idempotent overwrites are simpler and work when the aggregator emits a window's final count rather than deltas. See delivery semantics and idempotency.

Concrete settings and alternatives. Checkpoint every 30 to 60 seconds. With a transactional sink, output becomes visible only when a checkpoint commits, so end-to-end freshness is at least the checkpoint interval plus the watermark delay; with idempotent overwrites you can emit at window close and skip that wait. Allowed lateness of about 1 minute catches most late events; anything later goes to a side output and is corrected by the batch job. The alternative to stateful dedup is to count blindly and dedupe only in batch, which is simpler but makes streaming numbers overcount during retry storms; billing uses the batch result either way, so stream dedup is a freshness optimization and the batch dedup is the guarantee.

4. Hot ads and data skew

A viral ad can receive a large share of clicks, overloading the one partition that owns it.

• Key salting. Split the key into (ad_id, salt) with salt in 0..k, compute partial counts in parallel, then sum the k partial counts in a second stage. This removes the hot spot at the cost of an extra stage.
• Pre-aggregation (combiners). Each ingest node or first-stage operator counts locally over a short interval (for example, one second) and emits one partial count instead of many events. This reduces network traffic for all ads, hot or not.
• Scale by adding partitions. Requires repartitioning, which is expensive with checkpoints; plan partition counts ahead.

5. Querying

• Store per-minute counts. For longer ranges, either sum many minute rows or maintain rollup tables (hourly, daily) written by a downstream job. Rollups trade extra storage for faster range queries.
• Filtering by country or device adds dimensions to the key, multiplying rows; an OLAP store with columnar storage and pre-aggregation (Druid rollup) handles this natively.
• Top-N for the last M minutes is precomputed and stored per window, so queries are a single key lookup.

Query numbers. One ad for one day is 1,440 minute rows; a 30-day chart from minute rows is 43,200 rows, too many for a dashboard, so keep hourly (24 per day) and daily rollups and route by range (under 6 hours: minute rows; under 7 days: hourly; else daily). Cache top-N with a TTL of one window step (60 seconds), since it is recomputed that often anyway. Consistency model: reads are eventually consistent within the freshness bound; finalized billing numbers are only read after the reconciliation job marks a day closed.

Bottlenecks and scaling

• Ingest. Add queue partitions and consumers; partition by ad_id (or by click_id when skew makes ad_id unsuitable and the first stage only pre-aggregates).
• Aggregation state size. Keyed dedup state grows with click rate times TTL; bound it with TTL and store large state in an embedded disk-backed store (RocksDB in Flink).
• Aggregate DB writes. Batch writes; write per-window final counts rather than per-event increments.
• Backfill and reprocessing. Replay from raw storage into a separate pipeline and swap results, without touching the live path.
• Monitoring. Track consumer lag, watermark delay, dedup rate, and the difference between streaming and batch counts. A growing difference indicates data loss or double counting.

Failure modes

• Aggregator node dies. The job restarts from the last checkpoint and replays at most 30 to 60 seconds of input. Results are delayed, not wrong. Consumer lag and watermark delay rise and then drain at catch-up speed, which needs headroom of about 2x over peak.
• Aggregate database slow or down. The queue between aggregator and database absorbs writes (retain at least 24 to 72 hours); dashboards go stale but nothing is lost.
• Producer or region outage. Counts for that period are low, not duplicated. Reconciliation against raw storage and ad-server logs flags the gap.
• Poison or malformed events. Route to a dead-letter topic instead of failing the job; alert when the rate exceeds a small fraction of traffic.
• At 10x (500,000 clicks per second peak). Bytes are still only 50 MB per second, but dedup state grows to about 90 GB, partitions to 256 or more, and skew dominates: the top ads need salting by default and first-stage combiners become mandatory. Rollup and OLAP ingestion (not the queue) become the bottleneck, so pre-aggregate before the database.

Trade-offs and follow-up questions

• Why a queue between the aggregator and the database? It enables transactional exactly-once output and isolates database slowness from the stream job.
• Accuracy versus latency. Longer watermarks improve accuracy and delay results. Billing uses the reconciled batch numbers; dashboards use streaming numbers.
• Approximate counting. Sketches such as HyperLogLog and Count-Min Sketch reduce memory but are not acceptable for billing counts; they suit dashboards and unique-user counts.
• Click fraud. Bots inflate counts. A separate stage filters events (rate per IP or user, known bot signatures) before aggregation, and flagged events are excluded in reconciliation.
• What happens on a late event after the window has been emitted? It is either emitted as a correction that overwrites the window (idempotent key) or corrected by the nightly batch job.
• Follow-ups: unique clicks per ad (needs a distinct-count structure); attribution to conversions (joins with another stream); multi-region ingestion and aggregation.
• What if an advertiser disputes a count? Recompute from immutable raw events for that ad and window; the raw store is the audit trail.
• What if the dedup TTL is too short? Duplicates arriving after expiry are counted twice in streaming; the nightly batch dedup over the full day removes them.
• Can you get exactly-once without a transactional sink? Yes: emit final window counts under key (ad_id, minute) so replays overwrite rather than add.
• What if one ad gets 30% of traffic? Salt that key across about 10 sub-keys (detected by a heavy-hitter sketch or a config list) and sum in a second stage; other ads keep unsalted keys.