How to design an ad-click aggregation pipeline

Count ad clicks into one live dashboard number and one audited invoice number, from one pipeline read at two horizons, where a 0.1% miscount is real money.

The number that drives everything

• Traffic: 2.0e10 impressions/day, CTR 1.0%, CPC $0.45 → 200M clicks/day, ~$90M/day revenue.
• 0.1% accuracy target (per advertiser, at billing cut): 0.001 x 90M = $90,000/day, $32.9M/yr.
• Error is asymmetric: undercount = revenue never invoiced; overcount = refund + credit memo + audit (costs more). Concentration hurts: largest advertiser 10% of spend → $9,000/day exposure on one account.
• Clicks are 1% of traffic, 100% of the money → treat the two streams differently everywhere.

Two outputs, two guarantees

• Event time (stamped on device) not processing time: only event time makes replay bit-identical, the precondition for reconciliation. Processing time silently moves revenue between windows.
• Window: bucket of time counted together; every event lands in exactly one.

Watermark, priced

• Watermark w: promise “seen everything with event time < T”; a guess, price both sides.
• Skew tail: 92% arrive by 5 s, 98.5% by 30 s, 99.9% by 10 min, 99.999% needs 24 h.
• Drop cost = (1 - share(w)) x $90M; overshoot cost = w x $312.50/s (30% of spend caps out, overdelivery not billable). Total is U-shaped.
• Best single watermark ≈ 5 min → 0.30%, at least 3x over budget. No single w reaches 0.1%. This forces restatement.

Exactly-once, honestly

• End-to-end exactly-once delivery does not exist. What exists: at-least-once + idempotent write = effectively-once.
• Exact dedup on clicks only: event_id set, 24 h TTL → 200M x 24 B / 0.5 = 9.6 GB (fits one machine). Both streams = 970 GB (does not).
• Impressions: rotating ring of 24 hourly Bloom filters at p=1e-6 → 72.5 GB. Never on clicks: a bit array cannot enumerate which rows it dropped, and the auditor asks “which ones?”.
• Free second idempotence: upsert absolute cell value, never INCREMENT. Replaying a log slice is then a no-op; dedup only defends against duplicate input.

Lambda vs kappa: the pick

• Lambda (fast + batch serving layers): rejected. Two “click” definitions → diff dominated by your own codebase drift.
• Kappa (one stream path, replay): closer, but no single watermark hits 0.1%.
• Pick: kappa + restatement. One pipeline, two read horizons. Late events (past 30 s) go to a low-volume amendment stream that recomputes affected cells and upserts corrected values with a restatement record. Batch recount is an audit nobody queries.
• Result: pacing staleness 30 x 312.50 = $9,375/day + missing tail $900/day = $10,275/day = 0.011%, 8.8x inside budget. Win comes from closing early, not late.
• Cost: consumers must handle revision (as_of version), monotonicity breaks on IVT clawbacks (show restatements), replay must actually work (test weekly).

Hot partitions: salt then merge

• One advertiser = 10% of clicks. Consistent hashing balances keyspace, not a single key → still 0.10 + 0.90/64 = 7.30x skew on its partition. Virtual nodes do nothing.
• Salt the provably-hot key with S=16 (derived from event_id via a stable hash, not Python hash() which is per-process randomised) → 0.00625 + 0.90/64 = 1.30x. Only keys above uniform share (max 64) qualify; second merge stage sums the S partials (~512 rows/s, trivial).
• Stale salt map = one advertiser short by ~1/S in reconciliation.

Reconciliation: who wins

• Billed number is always the recount from the immutable log, never the stream. Log + 90-day archive (545 TB, RF 3) are system of record; snapshot is the only number anyone may invoice.
• Loop: T+26 h batch recount → reconciler diffs vs stream → inside threshold freeze + invoice; outside threshold hold + page on-call (wrong invoice costs more than a late one).
• Two thresholds: per-advertiser 0.05%, aggregate 0.01% (5x tighter). Independent noise shrinks by sqrt(n), so an aggregate hit can only be systematic (dropped partition, timezone, schema).
• Batch wins: bill the lower, then true up; never silently overwrite (log to restatements with reason); investigate the diff, not the total.
• Blind spot: both read the same log, so a collector that never sent an event is invisible to both → need an independent edge counter compared against receipts.
• IVT clawback (retroactive fraud removal) makes the amendment path a functional requirement, not an optimization.

                  ┌─ fast read (w=30s) → dashboards, pacing
log → dedup → aggregate → cells ┤
 │        (upsert absolute)      └─ final read (T+24h) → invoices
 │   late/IVT → amendment stream → cells
 └─ archive → T+26h batch recount → reconciler ─ in → billing snapshot
                                       └─ out → hold + page

Gotchas

• Money = integer micro-dollars. Reason is determinism, not precision: float addition is not associative, so two summation orders differ in low bits and the reconciler fires on rounding forever.
• Replay must reset both seen and cells: keep seen → empty window (undercount); keep cells → doubled window (overcount).
• Never resume with an empty dedup set → double-count (the expensive direction).
• Global watermark = min of per-partition watermarks; one stuck partition halts all windows → add per-partition idleness timeout.
• Reject events > 60 s in the future; carry NTP-disciplined event_time and received_time.