Design an Ad Click Aggregator

The synchronous increment has just fallen over on one viral ad's row. How do you take about 10k clicks a second without losing any — and still give the dashboard a number that moves?

The instinct is to keep the counter and spread it out: shard the count, add database nodes. For many problems that reflex is right. Here it is not, because the contention is not spread across the keyspace — it sits on one key. A viral ad's clicks all belong to one count, so its row is still one row no matter how many shards you add, and every write to it still serializes. Adding nodes helps the ads nobody is clicking. It does nothing for the one ad everybody is clicking. You cannot escape a hot row by adding cold ones. So the write path must not touch a shared counter at all.

Append every click to a durable, partitioned, retained event log, and count off to the side. Two clicks for the same ad are two independent appends — there is no row lock to fight over. The log is partitioned, so ingest scales by adding partitions. It is durable and replicated, so an accepted click cannot be lost. And the log is retained — kept for 30 days, not drained — because a second consumer, later in this article, will read the whole thing back. That is why it is a log and not a queue: a queue is drained and gone, and this record must survive to be replayed. For now the log is a shape with requirements, not a product; which real system fits it, and whether it can really take the load, is its own deep dive.

That takes the clicks. Now the fast number. A second service — the Live counter — reads the log as it fills and keeps the counts. The counting itself is simple. Take each click, look at its ad id, and add one to that ad's count for the current 1-minute window — a fixed slice of time that fills, is counted, and closes as the next one opens. (1 minute is the window Uber's real ad pipeline uses.) When a window closes, the counter writes each ad's count for that minute to the Results store, a database built for serving these small aggregate rows to the dashboard quickly. Because the Live counter reads from the log, it never touches a hot row: a viral ad is just more records. Because its counts live in memory, each click costs about 1 ms to process, not the 10 ms of a durable write. The number it serves is a few seconds behind and approximate — the next use cases say why — but it is fast. Like the log, the counter is a shape for now; the engine that runs it, and how it runs on more than one machine, is also its own deep dive. And the store holds only this provisional number; a second, exact number is the whole of the use case after.

The log keeps every click for 30 days. Put a size on that, starting with one click. A click record carries the fields from the brief's contract: the click id (a long random string, the biggest field), the ad id, the campaign id, a device or user id, a timestamp, and a small location tag — each a handful to a few dozen bytes, plus the field names and record framing around them. Call it 200 B per click, a stated planning size. Now the stream: 10k clicks a second × 200 B ≈ 2 MB/s of writing. Is that a lot? No — one ordinary machine writes sequentially to disk at roughly 200 MB/s, about a hundred times more. Writing was never the problem; the increment died on lock contention, not bandwidth. What makes the log heavy is keeping the stream: 2 MB/s that never stops is about 5.2TB over 30 days, and every byte is stored 3 times so a dead machine loses nothing — about 15.6TB of disk. No single machine holds that while also serving one writer and, later, two readers. So the log must be split across machines. And why keep 30 days at all? Because the log is only replayable as far back as it is kept. Billing must be able to re-derive any number it might still have to defend — advertisers are invoiced monthly and dispute bills after that — so the raw clicks are kept for a full billing cycle. Past that, the counts survive in the Results store; the raw clicks age out. The window is a choice with a price. One day of clicks is about 170 GB, and every byte has 3 copies, so each extra day of retention costs about 510 GB of real disk. Double the window and the log's storage doubles — and all the extra buys is the ability to recompute further back.

The unit of splitting is the partition: a slice of the log that is its own append-only file, living on its own machine, with 3 copies kept on other machines. The log here is split into 32 partitions — a stated planning choice — so each holds about 160 GB of the window. Every click is appended to exactly one partition. Which partition should a click land in?

The tempting partition key is the ad id: all of one ad's clicks in one partition, neatly together and in order. It fails for the reason this board keeps teaching. Ad popularity is extremely uneven, so one viral ad's partition takes a huge share of the stream while the rest sit idle — the hot row again, one layer down. A partition, like a row, is one file on one machine, and no number of other partitions helps the one being hammered.

Spread clicks evenly — hash the click id to pick the partition. Hash the click id — effectively random — so every partition receives an even share of the stream, viral ad or not. Adding capacity means adding partitions and machines. And nothing is lost by scattering an ad's clicks: the Live counter already regroups clicks by ad as it counts, so the log does not need to keep an ad's clicks together.

Now the shape's full requirement list is on the table: partitioned append-only files, replicated 3 ways, retained 30 days, an even spread of writes, a retried send dropped instead of stored twice, and the whole history replayable from any point. That list is a description of Kafka — a system built as exactly this: partitioned, replicated, retained logs. The retried send is handled by its idempotent producer: when our own ingest service re-sends a batch the log did not acknowledge, the batch's sequence number lets the log drop the copy. Note what that covers and what it does not: it protects the short hop between our ingest and the log. A phone re-sending a click from out in the world is a different duplicate — it arrives as a fresh request — and a later use case catches it by click id. So the design adopts Kafka here rather than rebuilding it. The replication stakes are worth naming too: the log is the one piece of this board that cannot be rebuilt if lost — everything else recomputes from it.

One more sizing worth naming: 32 partitions is far more than the 2 MB/s write rate needs. The count is chosen for the readers, not the writer — the Live counter's workers will read these partitions in parallel, and more partitions means more parallel readers. How that reading side scales is the next deep dive.

So far the Live counter has been one box, as if one machine counts everything. One machine can actually go a long way here — 10k events a second at about 1 ms of work each is a handful of busy cores, well within a single multi-core box's reach. But the box cannot stay one machine: peaks run several times the average, the ad system keeps growing, and a single counting machine is a single point of failure for every dashboard at once. How does the counter run on more than one machine, when its whole state is a set of running counts?

The tempting answer is to stay on one machine and buy a bigger one. It is simple, and for a while it works — the average load fits. It fails on three counts. Peaks run 3–5× the average — the brief's stated planning figure — and the biggest machine you can buy is a hard ceiling you meet at the worst moment. Growth compounds the same way. And one machine is one failure: when it dies mid-window, every ad's live count stops at once, and restarts are slow exactly when traffic is high. Scaling out needs an answer to one question first — who holds which ad's count — because two workers each holding half of one ad's count would answer wrong.

Parallel workers, split by ad id — each ad's count lives on exactly one worker. List what the counter actually needs. Run a small program — group clicks by ad, add to the current window's count — continuously over a stream, on several machines at once. Keep each ad's running count in exactly one place. Save that state regularly, and restore it when a machine dies. That list is the job description of a stream-processing engine, and Flink is the one this design adopts — the engine Uber's real ad pipeline runs. Flink runs the counting program as several identical workers and splits the stream between them by ad id: every click for a given ad is routed to the same worker, always. That worker alone holds that ad's running counts, in its own slice of state — what Flink calls keyed state. No count is ever split across two workers, so no merging is needed. The workers also share the reading: each reads some of the log's 32 partitions and routes each click onward to the worker that owns its ad. More partitions means more parallel readers — which is why the partition count was sized for readers.

Failure is handled the same way for one worker as for one machine before: every 2 minutes, each worker saves its slice of counts durably — the checkpoint. When a worker dies, its ads are reassigned to a replacement, which loads the last checkpoint and re-reads the log from that point. Nothing is lost, because the log kept the clicks; at most the dashboard's number stalls for a moment. A worker's state is small — its slice of the counts — so adding workers costs compute, not memory. One problem stays open. Splitting by ad id means one viral ad's entire stream still lands on the single worker that owns it, and adding workers does not help — the others cannot take clicks for an ad they do not own. That worker simply has to keep up, and past some click rate it cannot. The fix is to split that one ad's count into several sub-counts, so its clicks can spread across workers and be summed at the end. A later use case builds exactly that.

The fast number is back. But it is approximate, for three ordinary reasons — and the first two come from how a click travels. A click is sent by the person's phone or browser, and on a bad network the send can fail. The device keeps the click and sends it again when the connection returns, sometimes minutes later. So a click can arrive after its minute was already counted. And when the first send actually got through but the reply was lost, the device re-sends anyway — so the same click can arrive twice. Both copies carry the same click id, the one the device stamped when the click happened; that id is how a second copy can be recognized later. The third reason is on our side: a crash can lose counts that lived only in memory. Billing cannot bill from a number like that. So where does the exact number come from?

The tempting answer is to fix all three on the live path, then bill from it: wait longer before closing each minute so late clicks are included, remember seen click ids so a retry is not counted twice, and save the counts so often that a crash loses nothing. One number, one path, no second system. It fails, and the reason is structural, not a matter of effort. To be exact, a minute's count must wait for every click that belongs to it — including ones retried minutes later. A path that waits that long is no longer real-time. Complete means waiting; instant means not waiting. And even after giving up the speed, a crash can still lose whatever was in flight at that moment. Fast-and-exact from one path is not a hard problem. It is a contradiction. So the exact number has to come from a second path, one that is allowed to be slow. (A later use case does add guards for late and duplicated clicks — they make the fast number better, but they cannot make it exact.)

Keep the fast provisional count, and add a batch Reconciliation job that replays the log for the exact one. That second path is the batch reconciliation fork, the signature of this board. A Reconciliation job runs on a cadence — hourly, nightly — and replays the durable log from the start of the period. It reads the whole history at rest: nothing is in flight, every late click has landed, every duplicate can be resolved against the full record. So it can be exact. When it finishes, it overrides the provisional number in the Results store — an upsert on a stable key, the same ad-and-window row replaced in place. The Results store, so far just 'a serving database', can now be named too: it is Pinot, a database built for holding small aggregate rows like (ad, minute, count) and answering dashboard queries over them in milliseconds — and its upsert guarantees the store never keeps two records for the same identifier, which is exactly what the override needs. That is the standing rule of the fork: everything the fast path computes is eventually overridden by the batch path, by design. Two small costs ride along. During the override, old provisional rows and new exact rows briefly coexist, so the store needs about double the space for that window. And the store itself is never a bottleneck: a handful of dashboard and billing queries read what the whole firehose wrote, over tiny aggregate rows.

The cost is lag. The recompute runs on a cadence and takes time to finish, so the exact number settles a few hours behind — call it 2 to 6 hours. During that window, the only number for the most recent clicks is the provisional one, and the two numbers can legitimately disagree. That is not a bug. The fast number is a live guess. The exact number is the recomputed truth. The guess being a little off is the whole reason the recompute exists.

The Reconciliation job replays the log. Put a size on that too. A normal nightly run replays one day: about 170 GB. But the batch layer's real promise is bigger than a day. If a counting bug is found — say the seen-click-id check was mistakenly dropping real clicks for a week — the fix is to correct the code and recompute every affected day from the log. That is the whole reason the log keeps 30 days. And the worst case is recomputing all of it: the full 5.2TB window. How does one job replay that much log in time?

The tempting shape is one machine: the nightly day is about 170 GB, and at 200 MB/s sequential that reads in a quarter of an hour — one machine genuinely suffices for the normal night. It fails on the abnormal one. The batch layer exists precisely so a found bug can be recomputed away, and that means replaying not one day but every affected day — up to the whole 30-day, 5.2TB window. On one machine that is 7+ hours of pure reading before any counting, blowing past the 2–6 hour lag the design promised, exactly on the day trust in the numbers is lowest. Sizing the replay for the normal night and not the bad morning misses what the layer is for.

A distributed batch replay — Spark splits the read, shuffles partial counts, sums. The job runs on Spark, a batch engine whose whole trick is splitting a big read. The log is already in 32 partitions, so the split is natural: each of many workers reads some partitions and counts the clicks it sees, by ad, over data at rest. Note the difference from the live path: there, every click was routed to its ad's one owner before counting, so no merge was ever needed. The batch job does it the other way around — count first, merge after. Because clicks were spread evenly, every worker ends up holding partial counts for the same ads, so the workers exchange them: partial counts for each ad are sent to one place, grouped, and summed. Spark calls this exchange the shuffle. It is cheap: partial counts per ad are far smaller than the raw clicks they summarize, so the reading, not the exchange, dominates the run. The summed exact counts are then upserted into the Results store, overriding the provisional ones.

The payoff is that replay time divides by the worker count. The 7-hour single-machine read becomes under an hour on 10 workers, and the workers exist only while the job runs — a batch job can borrow a fleet for an hour in a way a live, always-on path never can. That is the quiet asymmetry of the fork: the live path must be provisioned for its worst second, while the batch path can rent its way through its worst day. A failed run just re-runs — the replay-and-upsert was already idempotent.

The two-path design works. But it left three problems open. A duplicated click from a re-sending phone still counts twice. A late click still has no rule for when a minute's count is done. And one viral ad still pins its one owning worker — the problem the counter's scaling left open. (The other cause of a skewed live count, a crash, the workers' checkpoints already cover.) The first two put a wrong number on the dashboard; the third stalls the counter. How do you close all three?

The first problem — duplicates — is the debate. A re-sent click reaches the Live counter as a second copy of the same click, carrying the same device-stamped click id, and counting it twice inflates the bill. So the counter keeps a set of recently seen click ids and skips any click whose id is already in it. (Flink stores this set the same way it stores the counts: in the job's own checkpointed state.) The question is the size of that set, because it is the biggest memory cost on this board. It must remember further back than the 1-minute window, because a retry can arrive minutes late — call it 1 hour, a stated planning choice. 1 hour at 10k clicks a second is about 36M ids. An exact set, at about 24 B an id, is about 864 MB, and it grows with the window and the rate. The alternative is a bloom filter: a probabilistic structure that answers definitely-never-seen or possibly-seen. At a 0.1% false-positive rate it holds the same window in about 65 MB — about 13× less. The cost: a false positive occasionally drops a real click as a duplicate.

A windowed bloom filter over the seen click ids — trade a tiny false-positive rate for far less memory. That dropped click is a small, bounded under-count, and the batch replay corrects it on the full log anyway — so the provisional live path can afford it. The exact set stays the fallback for when even a tunable error is unacceptable. Either way, the window is sized to the retry horizon and no wider: memory grows with the window, and a wider one only guards against very-late duplicates the batch replay already catches.

The second problem: late clicks. Every click carries the time it happened. But clicks do not arrive in that order — a click can be delayed by a slow network and show up after clicks that happened later. So when can the counter say a minute is finished and publish its count? It cannot wait forever. And it cannot close the minute the moment the clock passes it, because clicks from that minute are still on the way. Flink's rule for this is called a watermark. The counter tracks the newest click time it has seen so far, and subtracts a small waiting allowance — a few seconds. That moving mark is the watermark, and it means: everything up to this time has probably arrived. When the watermark passes the end of a minute, that minute's count is published. A click from that minute that arrives after publishing gets a grace period, called allowed-lateness: the count is updated and published again. Only a click later than the grace period is set aside — and even then the batch replay still counts it in the exact number. The allowance is a tuning knob: too tight drops real late clicks, too loose delays every minute's publish, and either way the batch replay is the backstop. One more rule: the watermark only moves when new clicks arrive. If one of the log's partitions goes quiet — a lull, a paused region — it would hold the whole watermark still. So a quiet partition is marked idle and skipped until it speaks again.

The third problem: the viral ad. All of its clicks go to the same counter key, on its one owning worker, so the update pressure lands on one key's state. A read hot key is fixed by serving copies from replicas. This is different — it is write skew. You cannot spread a write across replicas, because the point is to accumulate one count. So you salt the key: split the viral ad's counter into several sub-counters — the ad id plus a small salt — and sum them at the end. The clicks fan out, and no single key is pinned. This is forced, not optional: when one key takes a big share of the writes, adding machines cannot fix it, so you must spread the hot key itself.

One precision, because it is easy to conflate. Counting a click once is a chain of idempotency keys, not one guarantee: the log's producer drops a retried send at ingest, the dedup set drops a duplicate at aggregation, the store's upsert keeps the override from writing twice, and billing uses the click's own id so a click never bills twice. No single link is exactly-once by itself, which is why the Live counter needs its own dedup set on top of Kafka's. And this is count-once — count an event once inside an aggregate — not deliver-once, which is delivering a message once to a recipient. Similar keys, opposite failure domains.

Everything is built, and the happy path is complete. Now the premise arrives as a real incident — and it is not an off-to-the-side failure. It lands on the exact axis, the billable count, the one the whole reconciliation layer exists to protect.

The incident, plainly. Google's 2006 Lane's Gifts settlement paid up to $90M. $60M of it went back to advertisers as ad credits — about $4.50 refunded on every $1,000 spent, an overcharge of about 0.5%, across four years. And Google is not alone: in 2019, Facebook paid $40M over overstated video-watch metrics, $28M of it distributed to advertisers. Two dated, litigated incidents, both about a number that was supposed to be exact and was not. So — which number do you bill from, and is the batch layer worth its cost?

The tempting read is that the live count is close enough — it is usually within a fraction of a percent, so bill from it and save the second path. The incidents show why that is wrong. The provisional count drifts for ordinary reasons: a late click, a retry, a crash. Billed, each drift is money charged for clicks that did not happen. A fraction of a percent is not a rounding error once it touches a bill; it is a refund, a class action, a settlement. Billing cannot run on approximately right, because the cost of approximately is measured in litigation, not error bars.

Billing forbids a lossy path — keep the exact recompute, and never bill from the provisional count. The failure lives on the exact axis, exactly where the reconciliation fork was aimed. That fork is not an over-engineering luxury; it is what keeps a fractional error from becoming a settlement. It makes each piece load-bearing. The durable log must keep every click, because the recompute is only as complete as its record. The dedup must be exact or batch-corrected, so no click bills twice. And the exact number, never the provisional one, is what reaches billing. The fast number is for the dashboard, where close is fine. The exact number is for the bill, where close is a lawsuit.