Design an Ad Event Aggregator

An Ad Event Aggregator collects high-volume ad events (impressions, clicks, installs), enriches them with metadata, and serves both real-time dashboards and historical analytics queries. The same architecture applies to other high-volume metrics systems, but this walkthrough focuses on ad analytics. With 100M+ events per day, sub-second query latency, and strict data accuracy requirements, this question tests your understanding of stream processing, data enrichment, and OLAP database design.

This walkthrough follows the Interview Framework and focuses on what you'd actually present in a 45-60 minute interview.

Phase 1: Requirements

Functional Requirements

1. Collect events from multiple sources - Ingest click events, user behaviors, and metrics from web browsers, mobile apps, and backend services
2. Enrich events with metadata - Join raw events with dimensional data (user profiles, ad metadata, campaign info) to create enriched records
3. Real-time aggregation - Aggregate events in near real-time with minimum granularity of 1 minute
4. Query aggregated metrics - Advertisers/analysts can query metrics with sub-second response time
5. Support historical queries - Query data across time ranges from minutes to months

Non-Functional Requirements

• Scale: 10M active ads, peak of 10k events per second, ~100M events per day
• Latency: Sub-second query response time for analytics; events visible in dashboards within 1 minute
• Availability: 99.9% uptime - ads revenue depends on accurate data
• Durability: Zero data loss - every click represents revenue
• Idempotency: Same event should not be counted multiple times (exactly-once semantics)

Capacity Estimation

Traffic (Write - Events):

• Peak: 10k events per second
• Average: ~1,200 events per second (100M events / 86,400 seconds)
• Events are bursty—popular ads can spike to 100x normal traffic

Traffic (Read - Queries):

• Analytics queries: ~100-500 queries per second
• Dashboard refreshes: Every 10-60 seconds per active user
• Heavy queries (weekly/monthly aggregations) are less frequent but expensive

Storage:

• Raw event: ~500 bytes (event ID, timestamp, ad ID, user context, device info)
• 100M events/day × 500 bytes = ~50 GB/day raw events
• Retain raw events for 7-30 days = 350 GB - 1.5 TB
• Aggregated data is much smaller (~1-5 GB/day) but retained longer

Phase 2: Data Model

Core Entities

RawEvent (Ingested from clients)
├── event_id (PK, UUID or Snowflake ID)
├── event_type (click, impression, install, etc.)
├── ad_id
├── impression_id (unique per ad impression for idempotency)
├── timestamp
├── user_id (hashed or anonymized)
├── device_type (mobile, desktop, tablet)
├── platform (iOS, Android, Web)
├── geo_info (country, region)
├── referrer_url
└── client_timestamp

EnrichedEvent (After joining with metadata)
├── event_id (PK)
├── event_type
├── timestamp
├── ad_id
├── advertiser_id (from ad metadata)
├── campaign_id (from ad metadata)
├── ad_category (from ad metadata)
├── user_segment (from user profile)
├── device_type
├── platform
├── geo_info
└── enrichment_timestamp

AggregatedMetrics (Pre-computed for fast queries)
├── ad_id (PK component)
├── advertiser_id (PK component)
├── time_bucket (minute/hour/day - PK component)
├── granularity (enum: minute, hour, day)
├── click_count
├── impression_count
├── unique_users (HyperLogLog)
├── total_cost
└── last_updated

Advertisement (Dimensional data for enrichment)
├── ad_id (PK)
├── advertiser_id (FK)
├── campaign_id
├── redirect_url
├── category
├── cost_per_click
├── status (active, paused)
└── metadata (JSON)

Advertiser
├── advertiser_id (PK)
├── name
├── billing_info
└── settings

Database Selection

Event ID and Idempotency

Use impression_id as the idempotency key (assuming one click per impression):

• Each ad impression generates a unique impression_id on the client
• When a click occurs, it includes the impression_id
• Deduplication in the stream processor (keyed state + TTL) and/or OLAP dedupe (e.g., ReplacingMergeTree)
• This prevents counting the same click multiple times (e.g., user double-clicks, network retries)

If multiple clicks per impression are valid, keep impression_id for attribution but dedupe retries on a separate event_id.

Phase 3: API Design

Track Event (Click/Impression)

POST /api/v1/events
Headers: 
  X-Client-ID: <sdk_token>
  X-Request-ID: <idempotency_key>

Request Body:
{
  "event_type": "click",
  "impression_id": "imp_abc123",
  "ad_id": "ad_456",
  "timestamp": "2024-01-15T10:30:00.123Z",
  "device": {
    "type": "mobile",
    "platform": "iOS",
    "user_agent": "..."
  },
  "geo": {
    "country": "US",
    "region": "CA"
  }
}

Response: 202 Accepted
{
  "event_id": "evt_789",
  "status": "queued"
}

Use X-Request-ID to dedupe transport-level retries at ingestion; use impression_id for business-level idempotency in aggregation.

Click Redirect (For Ad Clicks)

GET /click/{impression_id}
Query Parameters:
  - ad_id: advertisement identifier
  - sig: HMAC signature for validation

Response: 302 Redirect
Location: https://advertiser.com/landing-page?utm_source=...

Side Effect: Enqueue click event for processing

Query Aggregated Metrics

GET /api/v1/metrics
Headers: Authorization: Bearer <token>

Query Parameters:
- advertiser_id: (required) filter by advertiser
- ad_ids: (optional) comma-separated list
- start_time: ISO8601 timestamp
- end_time: ISO8601 timestamp
- granularity: minute | hour | day
- group_by: ad_id, campaign_id, device_type, geo

Response: 200 OK
{
  "metrics": [
    {
      "ad_id": "ad_456",
      "time_bucket": "2024-01-15T10:00:00Z",
      "clicks": 1523,
      "impressions": 45000,
      "ctr": 0.0338,
      "unique_users": 1200,
      "total_cost": 152.30
    }
  ],
  "meta": {
    "granularity": "hour",
    "total_records": 24
  }
}

Batch Event Ingestion (For Mobile SDKs)

POST /api/v1/events/batch
Headers: 
  X-Client-ID: <sdk_token>
  Content-Encoding: gzip

Request Body:
{
  "events": [
    { "event_type": "impression", ... },
    { "event_type": "click", ... },
    ...
  ],
  "client_batch_id": "batch_123"
}

Response: 202 Accepted
{
  "accepted": 95,
  "rejected": 5,
  "errors": [
    { "index": 3, "error": "invalid_impression_id" }
  ]
}

Phase 4: High-Level Design

Architecture Diagram

Data Flow

1. Event Ingestion: Clients send events to the Load Balancer, which routes to the appropriate API (Click Processor for redirects, Event Collector for tracking)

2. Stream Buffering: Events are immediately written to Kafka/Kinesis for durability and decoupling

3. Stream Processing: Apache Flink consumes events, performs enrichment (joining with ad metadata), and computes real-time aggregations

4. Dual Write: Kafka Connect/Firehose sinks raw events to S3 (for batch reprocessing) while Flink writes aggregates to OLAP

5. Query Serving: Analytics Service reads from OLAP database, with Redis caching for hot queries

6. Batch Reconciliation: Periodic Spark jobs recompute aggregations from raw events in S3 to catch any streaming errors

Stream Processing with Flink

Apache Flink is ideal for this workload because:

• Exactly-once semantics: Prevents double-counting with checkpointing
• Windowed aggregations: Built-in support for tumbling/sliding windows
• Stateful processing: Maintains aggregation state in-memory
• Connectors: Native support for Kafka, S3, ClickHouse

// Simplified Flink aggregation job
DataStream<ClickEvent> clicks = env
    .addSource(new FlinkKafkaConsumer<>("clicks", ...))
    .keyBy(event -> event.getAdId())
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(new ClickAggregator())
    .addSink(new ClickHouseSink());

Phase 5: Deep Dives

Scaling to 10k Events Per Second

Let's walk through each bottleneck:

1. Click Processor API

• Stateless service, scales horizontally with load balancer
• Each instance handles ~2-3k requests/second
• Need 4-5 instances for 10k RPS with headroom

2. Kafka/Kinesis

• Both are distributed and can handle high throughput
• Shard by ad_id so all events for an ad go to the same partition
• Kinesis: 1 shard = 1 MB/s or 1000 records/s; need ~10-20 shards
• Kafka: Configure appropriate partition count based on consumer parallelism

3. Stream Processor (Flink)

• Scale by adding more task slots (parallelism)
• Each Flink job reads from one Kafka partition
• For 20 partitions, run 20 parallel tasks

4. OLAP Database

• Shard by advertiser_id (not ad_id) for query efficiency
• Advertisers query all their ads together
• ClickHouse can handle 100k+ inserts/second per node

Hot Shards Problem

Consider Nike launching a Super Bowl ad—that single ad_id gets 100x normal traffic, overwhelming its Kafka partition.

Solution: Salted Partition Keys

// Instead of partitioning only by ad_id
partitionKey = ad_id

// Use salted key for hot ads
partitionKey = ad_id + ":" + (hash(event_id) % N)
// Where N is the number of sub-partitions (e.g., 10)

For popular ads (determined by spend or historical volume), append a random suffix to distribute load across N partitions. The aggregation layer must then combine results from all N sub-partitions.

Data Enrichment Strategy

Raw events contain only IDs. Enrichment joins them with dimensional data:

Enrichment Data Sources:

• Ad metadata: ad_id → advertiser_id, campaign_id, category, CPC
• User profiles: user_id → user_segment, preferences
• Geo data: IP → country, region, city

Approach 1: Stream-Table Join (Recommended)

// Flink: Join event stream with a table in memory
EventStream
  .keyBy(event -> event.adId)
  .connect(adMetadataBroadcast)
  .process(new EnrichmentFunction())

• Ad metadata is broadcast to all Flink workers if it fits in memory
• Refresh broadcast table every 1-5 minutes
• Low latency, no external calls per event

Approach 2: Lookup Enrichment

// For each event, lookup metadata from cache/DB
EventStream
  .keyBy(...)
  .map(event -> {
    AdMetadata meta = cache.get(event.adId);
    return enrich(event, meta);
  })

• Uses Redis cache for lookups
• Higher latency, but handles larger datasets
• Good for user profile enrichment

Ensuring Zero Data Loss

1. Durable Message Queue

• Kafka/Kinesis replicate data across multiple brokers/shards
• Configure retention period: 7 days minimum
• If Flink fails, replay from the last checkpoint

2. Flink Checkpointing

• Checkpoint state to S3 every 1-5 minutes
• On failure, restart from the last checkpoint
• For 1-minute aggregation windows, checkpointing adds overhead

3. Dual Write to S3

• Kafka Connect/Firehose writes raw events to S3 (source of truth)
• Parquet format, partitioned by date/hour
• Enables batch reprocessing if streaming results are incorrect

Handling Late-Arriving Events

Mobile apps may send events hours or days after they occurred (e.g., user was offline). How do you handle late data?

Approach 1: Allowed Lateness Window

// Flink: Allow events up to 1 hour late
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.allowedLateness(Time.hours(1))

• Events within the lateness window trigger incremental updates to the aggregation
• Events beyond the window are dropped or sent to a side output for separate processing

Approach 2: Separate Late Events Table

• Late events (beyond 1 hour) go to a dedicated table
• Daily reconciliation job merges late events into main aggregations
• Dashboards show "preliminary" data with a disclaimer

Reconciliation: Balancing Speed and Accuracy

Stream processing is fast but can have errors (late events, out-of-order data, processing bugs). Batch processing is slow but more accurate.

Solution: Lambda Architecture (Simplified)

Reconciliation Process:

1. Flink writes aggregated data to OLAP (used for real-time dashboards)
2. Kafka Connect/Firehose writes raw events to S3
3. Daily Spark job re-aggregates from S3 (source of truth)
4. Compare Spark results with Flink results
5. If discrepancy > threshold, update OLAP with Spark values
6. Alert engineering team for investigation

Query Performance Optimization

Problem: Queries spanning weeks or months are slow even on OLAP databases.

Solution: Pre-aggregation at Multiple Granularities

-- Store aggregations at minute, hour, and day levels
CREATE TABLE aggregated_metrics (
    advertiser_id UInt64,
    ad_id UInt64,
    time_bucket DateTime,
    granularity Enum('minute', 'hour', 'day'),
    clicks UInt64,
    impressions UInt64,
    ...
) ENGINE = MergeTree()
PARTITION BY (granularity, toYYYYMM(time_bucket))
ORDER BY (advertiser_id, ad_id, time_bucket);

Rollup Strategy:

• Flink writes 1-minute aggregations in real-time
• Hourly cron job aggregates minutes → hours
• Daily cron job aggregates hours → days
• Queries for "last 30 days" read from day-level table (30 rows, not 43,200)

Trade-offs: Request Batching vs Latency

Mobile apps face a choice: send events immediately or batch them?

Recommendation: Hybrid

• Critical events (clicks): Send immediately for real-time attribution
• Non-critical events (impressions, scrolls): Batch every 30-60 seconds
• Offline queue: Store in SQLite, sync when online

// Mobile SDK pseudocode
function trackEvent(event) {
  if (event.type === 'click') {
    sendImmediately(event);
  } else {
    addToBatch(event);
  }
}

function addToBatch(event) {
  batch.push(event);
  if (batch.length >= 50 || timeSinceLastFlush > 30000) {
    flushBatch();
  }
}

Common Pitfalls

Interview Checklist

Before concluding, verify you've covered:

• Event ingestion with idempotency (impression_id)
• Stream processing architecture (Kafka → Flink → OLAP)
• Data enrichment strategy (broadcast join vs lookup)
• Real-time aggregation with windowing
• OLAP database for analytics queries
• Hot shard handling with salted partition keys
• Data durability (Kafka retention, S3 backup)
• Late-arriving events (allowed lateness, side outputs)
• Batch reconciliation for accuracy
• Pre-aggregation at multiple granularities
• Mobile SDK considerations (batching vs latency)
• Scaling each component horizontally
• Click redirect flow (server-side redirect)

Summary