Ad Click and Impression Aggregator

0) Problem Restatement

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.

1) Requirements

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

• Events: POST /v1/events (batched from ad servers and SDKs), mostly written straight to Kafka.
• Query: GET /v1/stats?campaign_id=77&from=...&to=...&granularity=hour → [{ time, impressions, clicks, ctr }]

2) High-Level Architecture

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

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 --> OL

3) Data 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)

4) Key Flows

1. The ad server sends an event with a unique event_id (created when the ad is served).
2. The collector writes it to Kafka and acknowledges. Kafka keeps it for 7 days.
3. 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).
4. When a minute is complete (after the watermark, explained below), Flink writes that minute's row to the OLAP store.
5. The nightly batch recounts the day from S3 and overwrites the stream numbers for that day. Billing uses only batch numbers.

5) Deep Dive A — Counting exactly once

Events can be duplicated (client retries, collector retries) or processed twice (a processor restarts).

• Deduplicate by event_id within a time window.
• Flink checkpoints: Flink saves its state and Kafka offsets together at regular intervals. After a crash, it restarts from the last checkpoint, so counts are neither lost nor doubled.
• Idempotent writes: the OLAP row for (ad_id, minute) is written as "set to X", not "add X". Writing the same row twice gives the same result.

6) Deep Dive B — Late events and hot ads

• 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.

7) Trade-offs & Alternatives

8) Common Follow-up Questions

• "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.

9) Wrap-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.