Design Real-Time Ad Click Event Aggregation

1. Problem Statement & Scope

System Mission

Design a mission-critical, fault-tolerant, real-time ad click event ingestion and aggregation pipeline capable of processing peak traffic of over 100,000 clicks per second ($1\text{ Billion clicks/day}$). The architecture must enforce strict Exactly-Once processing semantics for financial billing, execute multi-granularity tumbling and sliding window aggregations (1-minute, 1-hour, 24-hour CTR and spend) using Apache Flink, eliminate botnet and click-farm fraud via edge WAF and Redis sliding Bloom filters, and combine real-time stream aggregation with hourly EMR/Athena batch reconciliation (Lambda/Kappa architecture).

Synthesizing vector architecture diagram...

Functional Requirements

1. Ad Click Ingestion: Ingest and validate signed click tokens globally with sub-20ms response times and HTTP 302 advertiser redirect handoff.
2. Multi-Granularity Window Aggregation: Continuously compute click counts, valid clicks, bot/fraud clicks, and cumulative monetary spend partitioned by ad_id, campaign_id, and publisher_id across 1-minute tumbling, 1-hour tumbling, and 24-hour sliding windows.
3. Strict Exactly-Once Billing Semantics: Guarantee that advertisers are billed exactly once per valid click, even across network partitions, Flink worker crashes, and upstream message replays.
4. Real-Time Click Fraud Suppression: Filter out duplicate click IDs, rapid replay loops ($> 5\text{ clicks/min/IP}$), datacenter botnet traffic, and unverified token signatures before billing aggregation.
5. Advertiser Analytics Query API: Serve advertiser dashboards with sub-50ms query responses tracking Click-Through Rate (CTR) and real-time budget burn rate.
6. Lambda / Kappa Stream-Batch Reconciliation: Continuously stream raw Parquet events into an S3 data lake with hourly EMR/Athena batch jobs resolving late-arriving events and calculating billing adjustments.

Non-Functional Requirements (SLAs/SLOs)

• High Throughput: Support average $11,574\text{ clicks/sec}$, scaling to peak capacity of $\mathbf{100,000\text{ clicks/sec}}$ without data loss.
• Aggregation Freshness / Pipeline Latency: Aggregated metrics updated in advertiser dashboards within $< 3\text{ seconds}$ ($P99$).
• Data Correctness: $100\%$ financial billing accuracy with end-to-end exactly-once semantics. Zero double-billing under worker failover.
• Availability: $99.999\%$ uptime SLA for click ingestion endpoints ($< 5.26\text{ minutes}$ downtime/year).
• Fault Tolerance: Lossless Flink recovery via RocksDB checkpoints in S3 within $< 30\text{ seconds}$ of node crash.
• CAP / PACELC Classification:
  • Click Ingest Tier: AP / PA-EL (Availability and sub-20ms latency prioritized; immediate Kinesis buffer commit).
  • Billing Aggregation & Flink State: CP / PC-EC (Strict consistency via RocksDB checkpoints, idempotent transactional DynamoDB sinks).

---

2. Capacity & Scale Estimation

Traffic Calculations

• Daily Ad Impressions: $50\text{ Billion}$ impressions/day.
• Average Click-Through Rate (CTR): $2\% \implies 1\text{ Billion clicks/day}$.
• Click Ingestion Throughput: $\text{Avg Ingestion QPS} = \frac{10^9\text{ clicks}}{86,400\text{ sec}} \approx \mathbf{11,574\text{ clicks/sec}}$ $\text{Peak Ingestion QPS (5}\times\text{ Black Friday / Super Bowl surge)} = \mathbf{100,000\text{ clicks/sec}}$
• Query Load (Advertiser Dashboards):
  • 100,000 active campaigns queried every 10 seconds by advertiser automation bots. $\text{Analytics Query QPS} = \frac{100,000}{10\text{ sec}} = \mathbf{10,000\text{ QPS (Peak: 25,000 QPS)}}$

Storage & Data Sizing Math

• Raw Click Event Payload: $\approx 500\text{ bytes}$ (Protobuf/JSON: click_id, ad_id, campaign_id, publisher_id, user_id, ip_hash, cpc_cents, event_timestamp, token_signature). $\text{Peak Ingress Network Bandwidth} = 100,000\text{ clicks/s} \times 500\text{ bytes} = 50\text{ MB/s} = \mathbf{400\text{ Mbps}}$ $\text{Daily Raw Ingestion Volume} = 10^9 \times 500\text{ bytes} = \mathbf{500\text{ GB/day}}$
• Columnar Snappy Parquet Data Lake (S3):
  • Snappy Parquet achieves $4\times$ compression over raw JSON: $\text{Compressed S3 Storage} = \frac{500\text{ GB}}{4} = \mathbf{125\text{ GB/day}} \implies \mathbf{45.6\text{ TB/year}}$ 3-year data lake retention: $45.6 \times 3 \approx \mathbf{136.8\text{ TB}}$.
• Aggregated Metric Record Volume (Write Reduction):
  • 100,000 active ads aggregated across 1-minute tumbling windows: $\text{Aggregated Writes to DynamoDB} = \frac{100,000\text{ rows}}{60\text{ sec}} \approx \mathbf{1,667\text{ writes/sec}}$
  • Write Load Reduction: Compared to writing raw clicks ($100,000\text{ WPS}$), stream aggregation slashes database write IOPS from $100,000 \to 1,667$, achieving a $98.3\%$ write load reduction.

Memory & Cache Sizing (80/20 Pareto Working Set)

• Redis Sliding Bloom Filter (Duplicate Click Suppression):
  • Sliding window of 10 minutes ($600\text{ seconds}$).
  • Number of clicks in 10 minutes at peak: $n = 100,000 \times 600 = 60\text{ Million clicks}$.
  • Target false positive probability $p = 0.001$ ($0.1\%$): $m = - \frac{n \ln p}{(\ln 2)^2} = - \frac{60 \times 10^6 \times \ln(0.001)}{(0.69315)^2} \approx 862.6 \times 10^6\text{ bits} \approx \mathbf{107.8\text{ MB RAM}}$
  • Double-buffered rotating Bloom filter (Current + Previous window): $108\text{ MB} \times 2 \approx \mathbf{216\text{ MB RAM}}$.
• IP Velocity Rate Limiting (ElastiCache Redis):
  • Top 10 Million active IPs tracked via sliding token bucket in Redis: $10 \times 10^6 \times 64\text{ bytes} \approx \mathbf{640\text{ MB RAM}}$
• Flink RocksDB Stateful Working Set:
  • In-memory window state for 100,000 campaigns across 1-minute and 1-hour active windows: $100,000 \times 1\text{ KB} \approx \mathbf{100\text{ MB state per Flink worker}}$

Fleet Sizing & Kinesis Provisioning

• Amazon Kinesis Data Streams Provisioning:
  • 1 Kinesis shard provides $1\text{ MB/sec}$ ingress or $1,000\text{ records/sec}$.
  • Required shards for $100,000\text{ records/sec}$ and $50\text{ MB/sec}$: $\text{Shard Count} = \max\left( \frac{100,000}{1,000}, \frac{50\text{ MB/s}}{1\text{ MB/s}} \right) = 100\text{ Shards} \to \mathbf{128\text{ Shards (Power of 2 for hashing)}}$
• Apache Flink Compute Sizing (AWS Kinesis Analytics):
  • Sized at 32 Kinesis Processing Units (KPUs, each with 1 vCPU and 4 GB RAM).
  • Parallelism $= 128$ (aligned 1:1 with Kinesis shards).

---

3. AWS-First High-Level Architecture

Synthesizing vector architecture diagram...

Follow a click through the hot path and then the cold path. In the "Stateless High-Throughput Ingestion Fleet" panel, the NLB spreads clicks across ingest tasks, which write to Kinesis partitioned by hash(ad_id + salt), so one viral ad is spread over several shards. In the "Stateful Stream Processing" panel, Flink groups clicks by the time they happened (event time), waits for late arrivals up to a watermark, drops duplicates and suspicious IPs using Redis, checkpoints its state to S3 every 30 s, and writes per-minute totals to DynamoDB. In the "Cold Path Data Lake & Batch Reconciliation" panel, the same raw clicks are also written to S3 by hour, and an hourly Spark job recomputes the totals, correcting DynamoDB for clicks that arrived too late. Advertisers read the totals through the "Advertiser Dashboard Serving Tier" panel. The fast path gives near-real-time numbers; the batch path makes the billed numbers exact.

Data Flow Walkthrough

1. Ingest & Instant Redirect: The user clicks an ad. The browser hits the NLB. The ECS Go Ingestion worker validates the HMAC impression token signature, dispatches the click event to Kinesis, and immediately issues an HTTP 302 redirect to the advertiser landing page in $< 15\text{ ms}$.
2. Kinesis Partition Salting: To prevent mega-campaigns (e.g. Super Bowl ads) from overwhelming a single Kinesis shard, the partition key is salted: ad_id + "#" + random(0, 63), spreading writes across up to 64 shards. The salt range is sized from the hottest campaign: $40{,}000$ clicks/sec on one ad divided by the $1{,}000$ records/sec shard limit needs at least 40 salts; 16 would still leave $2{,}500$/sec per shard and throttle.
3. Stateful Flink Aggregation: Apache Flink consumes Kinesis events using Event-Time watermarks. It checks ElastiCache Redis Bloom filters to drop duplicate click IDs and verifies IP velocity limiters. Valid clicks increment tumbling 1-minute and 1-hour window aggregates in RocksDB.
4. Idempotent DynamoDB Sink: DynamoDB is not a transactional participant in Flink's two-phase commit, so the sink must be idempotent instead: when a 1-minute window closes, Flink writes the window's absolute totals to the row keyed by (ad_id, window_start) with SET valid_clicks = :v, spend_cents = :s, checkpoint_id = :cid and the condition attribute_not_exists(checkpoint_id) OR checkpoint_id < :cid. A replay after a crash rewrites the same totals for the same window and is a no-op. Using ADD valid_clicks :inc here is the classic double-billing bug: every replayed flush would add the window again.
5. Cold-Path Reconciliation: Firehose streams raw events to S3 in Snappy Parquet. An hourly Amazon EMR job computes ground-truth aggregates over the raw data lake, catching late-arriving clicks beyond Flink's 5-minute watermark and writing financial reconciliations.

Concrete Step-by-Step Request Walkthrough: Tracing Click Ingest to Billing Update

---

4. API Interface Design

1. Ingest Ad Click Event (POST /v1/events/adclick)

Receives incoming click beacons, validates cryptographic signatures, and outputs redirect instructions.

httpCopy
POST /v1/events/adclick HTTP/1.1
Host: click.ads.aws.internal
User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 17_4 like Mac OS X)...
X-Forwarded-For: 203.0.113.195
Content-Type: application/json

{
  "click_id": "clk_88a91c74f0b21a",
  "ad_id": "ad_nike_airmax_2026",
  "campaign_id": "camp_nike_spring_global",
  "publisher_id": "pub_techcrunch_01",
  "user_id": "usr_998124a87",
  "cpc_cents": 45,
  "impression_timestamp": 1773648000100,
  "click_timestamp": 1773648002340,
  "click_token_sig": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
}

Response: 202 Accepted

jsonCopy
{
  "status": "ACCEPTED",
  "redirect_url": "https://www.nike.com/air-max-2026?utm_source=ad_campaign",
  "ingest_timestamp": 1773648002352
}

2. Query Real-Time Campaign Performance (GET /v1/analytics/campaigns/{id}/metrics)

Returns pre-aggregated campaign performance metrics across specified time windows.

httpCopy
GET /v1/analytics/campaigns/camp_nike_spring_global/metrics?window=1h&start_time=1773648000&end_time=1773651600 HTTP/1.1
Host: dashboard.ads.aws.internal
Authorization: Bearer <jwt_token>

Response: 200 OK

jsonCopy
{
  "campaign_id": "camp_nike_spring_global",
  "window": "1_HOUR",
  "start_time": 1773648000,
  "end_time": 1773651600,
  "metrics": {
    "total_impressions": 5000000,
    "total_clicks": 142850,
    "valid_clicks": 139210,
    "fraud_clicks_blocked": 3640,
    "total_spend_cents": 6264450,
    "ctr_percentage": 2.78,
    "average_cpc_cents": 45.0
  }
}

3. Status Codes & Error Contracts

---

5. Data Models & Storage Architecture

Relational Schema (Amazon Aurora / Redshift Cold Path Analytics)

sqlCopy
CREATE TABLE ad_clicks_raw (
    click_id VARCHAR(64) PRIMARY KEY,
    ad_id VARCHAR(64) NOT NULL,
    campaign_id VARCHAR(64) NOT NULL,
    publisher_id VARCHAR(64) NOT NULL,
    user_id VARCHAR(64),
    ip_hash CHAR(64) NOT NULL,
    cpc_cents INT NOT NULL,
    is_valid BOOLEAN NOT NULL DEFAULT TRUE,
    fraud_reason VARCHAR(32),
    event_timestamp TIMESTAMPTZ NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
) PARTITION BY RANGE (event_timestamp);

CREATE TABLE campaign_hourly_aggregates (
    campaign_id VARCHAR(64) NOT NULL,
    window_start TIMESTAMPTZ NOT NULL,
    total_clicks BIGINT NOT NULL DEFAULT 0,
    valid_clicks BIGINT NOT NULL DEFAULT 0,
    fraud_clicks BIGINT NOT NULL DEFAULT 0,
    total_spend_cents BIGINT NOT NULL DEFAULT 0,
    reconciled_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (campaign_id, window_start)
);

DynamoDB Single-Table Schema (AdAggregatesTable)

---