Ad Click Event Aggregation

Design an ad click event aggregation system at Facebook/Google scale. Ad click aggregation measures advertising effectiveness: it impacts billing, campaign budgets, bidding strategies. Key metrics: CTR (click-through rate), CVR (conversion rate).

RTB (Real-Time Bidding): the core process where digital ad inventory is bought/sold in <1 second. RTB latency is sub-second; aggregation latency can be minutes.

Step 1 of 4: Understand the problem and establish design scope #

Goal: lock in requirements and scale.

Functional requirements #

Must have:

1. Aggregate click count for a given ad_id in the last M minutes
2. Return top 100 most clicked ads in the past 1 minute (both parameters configurable). Aggregation every minute
3. Support filtering by ip, user_id, or country
4. Input: log files on different servers. Fields: ad_id, click_timestamp, user_id, ip, country
5. Handle: late-arriving events, duplicated events, system recovery

Non-functional:

1. Correctness critical (used for billing)
2. Handle delayed/duplicate events
3. Robust against partial failures
4. End-to-end latency: a few minutes max

Back-of-the-envelope estimation:

1. 1B DAU × 1 ad click/user/day = 1B ad click events/day
2. Ad click QPS = 10⁹ / 10⁵ = ~10,000 (avg). Peak = 50,000 QPS (5×)
3. Storage: 0.1KB/event × 1B = 100 GB/day, ~3 TB/month
4. 2M total ads. 30% YoY growth → doubles every ~3 years

Step 2 of 4: Propose high-level design and get buy-in #

Goal: agree on the query APIs, data model, and two-queue pipeline.

Query API design #

Clients are dashboard users (data scientists, product managers, advertisers).

API 1: Aggregate click count

Parameters: from (start minute), to (end minute), filter (filtering strategy ID). Response: {ad_id, count}.

API 2: Top N most clicked ads

Parameters: count (top N), window (M minutes), filter. Response: {ad_ids: array}.

Data model #

Raw data (log files):

Aggregated data (per minute):

With filters (star schema):

Filter table maps filter_id → {region, IP, user_id}.

Top N structure:

Store both raw and aggregated data:

Recommendation: store both. Raw = backup + debug + recalculation source (move old to cold storage). Aggregated = active query data.

Database choice #

1. Raw data: write-heavy (10K avg QPS, 50K peak), low read volume. Cassandra or InfluxDB (optimized for write + time-range queries). Alternative: columnar formats (ORC, Parquet, AVRO) on S3 with file rotation
2. Aggregated data: read-heavy (2M ads queried every minute for dashboards) AND write-heavy (every-minute aggregation). Same DB type works for both

High-level design #

Unbounded data stream → aggregation → aggregated results. Asynchronous processing via two Kafka queues:

Message queue 1: raw ad click events (ad_id, click_timestamp, user_id, ip, country).

Message queue 2: two data types per minute:

1. Ad click counts (ad_id, click_minute, count)
2. Top N most clicked ads (update_time_minute, most_clicked_ads)

The second Kafka is needed for end-to-end exactly-once semantics (atomic commit).

Aggregation service (MapReduce DAG) #

The DAG model breaks the system into small computing units:

1. Map node: reads from the data source, filters, transforms. Routes by ad_id % N to aggregate nodes. Needed for: data cleaning/normalization, handling events of the same ad_id landing in different Kafka partitions

2. Aggregate node: counts ad click events by ad_id in memory every minute
3. Reduce node: reduces results from all aggregate nodes to the final result. Example: 3 aggregate nodes each have a top 3 → Reduce node merges to the top 3 overall

DAG = MapReduce paradigm. Intermediate data in memory; nodes communicate via TCP (different processes) or shared memory (different threads).

Use case 1: Aggregate click count

Input events partitioned by ad_id % 3 in Map nodes → aggregated by Aggregate nodes.

Use case 2: Top N most clicked ads

Each Aggregate node maintains a heap for efficient top-k. Reduce node merges 3N candidates → final top N.

Use case 3: Data filtering

Pre-define filtering criteria; aggregate with dimensions. Star schema: filtering fields = dimensions. Benefits: simple, reuses the aggregation service, fast pre-calculated access. Limitation: many buckets/records with many filter criteria.

Step 3 of 4: Design deep dive #

Goal: architecture style, time handling, delivery guarantees, scaling, fault tolerance.

Streaming vs batching #

The system uses both: streaming for real-time aggregation, batch for historical backup.

Lambda architecture: two processing paths (batch + streaming). Con: two codebases. Kappa architecture: a single stream processing engine for both. Our design uses Kappa: historical replay goes through the same aggregation service.

Data recalculation (historical replay):

1. Recalculation service retrieves raw data (batch job)
2. Sends to a dedicated aggregation service (doesn't impact real-time processing)
3. Aggregated results → second Kafka → aggregation database

Time #

1. Event time: when the ad click happened. More accurate, but depends on the client clock (may be wrong/malicious)
2. Processing time: server system time. More reliable, but inaccurate for delayed events

Recommendation: event time (data accuracy is critical for billing).

Watermark technique:

An extension of the aggregation window to catch slightly delayed events. Example: 15s watermark extension. Tradeoff: longer watermark → more accurate but higher latency. Does NOT handle long-delayed events (low ROI for complex design). Minor inaccuracies are corrected by end-of-day reconciliation.

Aggregation window #

1. Tumbling window (fixed): equal-length, non-overlapping chunks. Used for per-minute click count aggregation

2. Sliding window: overlapping windows sliding by a specified interval. Used for "top N in last M minutes"

Delivery guarantees #

At-least-once (small % duplicates acceptable) is NOT sufficient: discrepancies of millions of dollars. Exactly-once required.

Data deduplication: sources of duplicates:

1. Client-side: malicious/intentional resends (handled by ad fraud/risk control)
2. Server outage: aggregator crashes mid-aggregation; upstream offset not updated → new aggregator replays already-processed events

Solution: distributed transaction. Save the offset AND send results atomically. If any step fails, the entire transaction rolls back.

Scale the system #

Three decoupled components → scale independently.

Message queue:

1. Producers: unlimited instances
2. Consumers: consumer group rebalancing. Add during off-peak (rebalance is slow with hundreds of consumers)

3. Brokers:
   • Hash key = ad_id → same ad_id events in the same partition
   • Pre-allocate enough partitions (avoid dynamic increase)
   • Topic physical sharding: by geography (topic_north_america, etc.) or business type. Increases throughput; adds complexity

Aggregation service:

Horizontally scalable. Two options for throughput:

1. Multi-threading (events by ad_id → different threads)
2. Resource providers (YARN) → multi-processing. More common in practice

Database:

Cassandra natively supports horizontal scaling via virtual nodes + consistent hashing. Data evenly distributed with replication factor. Adding a node → auto-rebalances. No manual resharding.

Hotspot issue #

Popular ads (major advertisers with million-dollar budgets) get more clicks. Since events are partitioned by ad_id, some aggregation nodes overload.

Mitigation: allocate more aggregation nodes for popular ads via the resource manager. The original node splits events → multiple nodes process → results written back.

More sophisticated: Global-Local Aggregation, Split Distinct Aggregation.

Fault tolerance #

Aggregation is in-memory → a node crash loses results. Rebuild by replaying from upstream Kafka. Use snapshots (system state + upstream offset + top N data) to avoid replaying from the beginning.

Failover: bring up a new node → recover from the latest snapshot → replay events after the snapshot from Kafka.

Data monitoring and correctness #

Continuous monitoring:

1. Latency: track timestamps through each stage; differences = latency metrics
2. Message queue size: sudden increase → add aggregation nodes. For Kafka: monitor records-lag
3. Aggregation node resources: CPU, disk, JVM

Reconciliation:

End-of-day batch job: sort events by event time in each partition → compare with the real-time aggregation result. May not match exactly (late events). For higher accuracy: use hourly windows.

Alternative design #

Store ad click data in Hive with an ElasticSearch layer for fast queries. Aggregation in OLAP databases (ClickHouse, Druid).

Step 4 of 4: Wrap up #

What we designed:

1. MapReduce DAG for aggregation
2. Kappa architecture (streaming + batch replay combined)
3. Watermark for delayed events
4. Distributed transactions for exactly-once
5. Consistent hashing for Cassandra scaling
6. Hotspot mitigation via resource manager
7. Snapshot-based fault tolerance
8. Reconciliation for correctness