22 — Ad Click Event Aggregation

“Ingest ad click events and serve clicks(ad_id, minute) and top_k(advertiser, window) — accurately enough to bill on.”

An ad click aggregation system counts events — somebody clicked an advertisement — and turns those counts into two things: a live dashboard number and an invoice.

By the end of the chapter you should be able to:

• Say precisely what a window is, and why the timestamp written on the event beats the clock on the server that receives it.
• Price a watermark — the pipeline’s decision about how long to wait for stragglers — in dollars per day rather than in taste.
• Explain what “exactly-once” can and cannot mean once money is involved.
• Describe the reconciliation loop that decides which of two disagreeing numbers gets billed.

What goes in, and what comes out

In goes a stream of small events, roughly 100 bytes each, produced by browsers and mobile apps all over the world. Each one says: this ad, this user, this country, at this instant, worth this much money.

Out come two different artifacts, and keeping them straight is most of the design.

The first is clicks(ad_id, minute) — a count per advertisement per minute, refreshed within seconds. That is what a campaign manager watches on a dashboard and what the budget-pacing loop reads to decide whether to keep serving an ad.

The second is a billing snapshot: one immutable, audited number per advertiser per day, produced a day later. That is the number multiplied by a price and mailed out as an invoice.

Those two outputs have different freshness requirements and different accuracy requirements. The whole design comes from one move: they are produced by one pipeline, read at two different moments.

Why this one cannot be “eventually roughly right”

Every other aggregation system in this book is allowed to be eventually roughly right. This one is not, because the number it produces is multiplied by a price and mailed to a customer.

The whole design rests on one sentence: an analytics pipeline that loses 1% of its events has a rounding error, and a billing pipeline that loses 0.01 x 90,000,000 = 900,000 dollars a day has a lawsuit.

Ownership split with the ML chapter

Two chapters in this book touch ad clicks, and they own different halves.

ml/08 — Ad Click Prediction owns prediction: the calibrated pCTR model, the auction that consumes it, position bias, and delayed conversions. pCTR is predicted click-through rate — the model’s estimate of the probability that a given user clicks a given ad, computed before the ad is shown.

This chapter owns the counting pipeline that runs after the click happens: ingestion, windowed aggregation, exactly-once, late and out-of-order events, and reconciliation against billing.

The two chapters share exactly one thing, the traffic model, which is stated there and cited here. You can read this chapter without opening that one.

The argument, in six steps

Six claims carry the argument, each settled by one number; the bold rows are the result the design turns on.

1. Framing: what decision, and what breaks

A click happens on a device you do not control. It travels through a network that drops and duplicates. It lands in a system that must eventually agree with an invoice. Four decisions follow from that sentence:

1. What a window means. A window is the bucket of time whose events are counted together — for example, “all clicks in the minute beginning 10:03”. Every event has to land in exactly one bucket, and the two computations of a bucket’s contents have to agree.
2. How long you wait for stragglers. That is the watermark question, and The watermark priced prices it.
3. What “exactly once” is allowed to mean, given that the network guarantees nothing.
4. Who wins when the fast streaming number and the slow batch number disagree.

Three things can break. The table lists them in increasing order of how much they cost you, and only the third one is interesting — the first two are ordinary latency problems.

Requirements

Functional

• ingest(events) — accept click and impression events (an impression is one showing of an ad, whether or not anybody clicked it) from edge collectors, the small servers placed close to users that receive events and forward them. Delivery is at-least-once: the network may deliver an event more than once, and never silently drops one.
• `query(ad_id | campaign_id | advertiser_id, Reconciliation against billing and who wins is why.
• Filter invalid traffic (IVT) — clicks generated by bots or fraud rather than people — before billing, retroactively, without losing the count as it stood before the filter ran.

Non-functional

These are the numbers every later section is measured against, and the accuracy target is the unusual one — the next subsection derives it.

Four terms in that table are used from here to the end of the chapter. Fix them now:

• T+24 h — twenty-four hours after the end of the day being billed, where T is that day. The billing cut is therefore a full day behind the dashboard, on purpose.
• Event time — the instant the click happened, as stamped on the device where it happened. Its opposite is processing time: whenever the pipeline got round to looking at the event.
• Late arrival — an event that reaches the pipeline after the window it belongs to has already been closed and published. A phone that was in a tunnel; an app that batched events while offline and flushed them on reconnect.
• Immutable — written once and never altered. The raw log is append-only, so a recount performed tomorrow reads exactly the same bytes today’s count read.

Note the shape of that table: ingest availability and billing accuracy are hard; query latency is not. That is backwards from most systems in this book, and saying it out loud early is worth a lot.

Back-of-envelope: what a counting error costs

One number justifies the entire design: what a tenth of a percent of miscounting is worth. Price that first, then size the traffic, the events, and the storage against it.

The traffic model, and the price of being wrong by 0.1%

The traffic model comes from Framing. Three inputs:

• 2.0e10 ad impressions per day. 2.0e10 is scientific notation for 20 billion.
• A click-through rate (CTR) of 1.0% — the fraction of impressions that get clicked.
• An average cost per click (CPC) of $0.45 — what the advertiser pays for each click.

Do not re-derive that model; take it and price the error. Four lines, each one substituting the line above it:

clicks/day             20,000,000,000 x 0.01     =  200,000,000
click revenue/day      200,000,000 x 0.45        =   90,000,000     $
0.1 % of a day         0.001 x 90,000,000        =       90,000     $
annualized             90,000 x 365              =   32,850,000     $

A tenth of a percent of miscounting is a $32.9M/yr line item.

The error is also asymmetric in a way that matters. Undercounting is revenue you never invoice — you lose the money and nobody outside finance finds out. Overcounting is revenue you invoice and then refund, with an apology, a credit memo, and an audit. The second one costs more than the money.

Concentration makes it worse still, because errors do not average out per customer. One large advertiser is a single person in a single meeting looking at a single number:

largest advertiser's share of spend                =        0.10
their spend/day        0.10 x 90,000,000           =   9,000,000    $
0.1 % of that          0.001 x 9,000,000           =       9,000    $/day

One account, one meeting, $9,000/day. That is the conversation the design has to survive.

Rates: what the pipeline handles per second

Now the rates and volumes (The three numbers you actually need has the templates). Each line divides a per-day figure by 86,400 — the number of seconds in a day — to get a per-second average, then multiplies by 3.

peak x3 means the busiest hour of the day runs at about three times the daily average. The peak column, not the average column, is the number you provision machines for.

impressions/s   20,000,000,000 / 86,400   =  231,481      peak x3  =  694,443
clicks/s           200,000,000 / 86,400   =    2,315      peak x3  =    6,945
events/s        20,200,000,000 / 86,400   =  233,796      peak x3  =  701,389

Clicks are 1% of the traffic and 100% of the money. That single ratio justifies treating the two streams differently everywhere below — different dedup, different retention, different accuracy budget.

Reads: how many dashboard queries per second

The read side needs a rate too. It is the one workload figure nothing upstream measures, so build it from a stated assumption rather than asserting a number. The assumption has three parts:

• 100,000 advertisers — the same count Reconciliation against billing and who wins uses for its independence argument.
• 2% of them have a dashboard open during the peak hour.
• Each open dashboard refreshes once a second, matching the 1-second staleness budget from Requirements.

dashboards open at peak   100,000 x 0.02   =   2,000
dashboard queries/s       2,000 x 1/s      =   2,000

2,000 queries/s of reads against 6,945 clicks/s of writes at peak — clicks, not events.

Say which stream you mean, because the two differ by a hundred times. The pipeline ingests 701,389 events/s at peak; reads are the smaller number only against the billable stream of 6,945 clicks/s. Reads do not outnumber the full firehose and never will.

Reads being the smaller of the two numbers that matter is what lets Windowing and why event time is not a preference spend read work to buy back write work.

The 2% is the softest figure in the chapter, and the assumption ledger records it as such. Halve it or double it and nothing in the design changes shape, because the query tier sits behind a 1-second cache either way.

Event size, bandwidth, and retention

Next, how big one event is. Here is the field list for a single click record, with each field’s size in bytes:

event_id      (uuid, binary)                16
event_time    (epoch microseconds)           8
ad_id                                        8
campaign_id                                  8
advertiser_id                                8
user_id       (hashed)                       8
placement_id                                 8
price_micros  (integer micro-dollars)        8
country                                      2
device_type                                  1
ip_prefix_hash                               4
                                            ---
                                             79

The fields add up to 79 bytes. Round to 100 B on the wire once framing overhead and a compact binary encoding are included.

Now the ingest bandwidth. RF 3 in the block below is a replication factor of three: the log keeps every event on three separate machines, so losing one machine loses nothing. Replication multiplies the bytes you must move and store by 3.

events/s        231,481 + 2,315           =    233,796
bytes/s         233,796 x 100             = 23,379,600      = 23.4 MB/s
with log RF 3   23,379,600 x 3            = 70,138,800      = 70.1 MB/s

Compare 70.1 MB/s against one machine’s network capacity. The standing assumption in this book is 125 MB/s per machine — one gigabit per second through a network interface card (NIC), the hardware that connects a machine to the network.

So the entire firehose fits on one NIC and does not fit on one NIC. Arithmetically 70.1 < 125, so it would fit. You split it across many machines anyway, for two reasons:

• One machine failing should cost you a fraction of the stream, not all of it.
• Hot partitions salt then merge cannot fix hot keys until the traffic is already divided into partitions — independent slices of the stream, each handled by its own machine.

Last, raw retention: how much of the immutable log you keep, and for how long. Both streams together this time, not just clicks.

raw bytes/day   20,200,000,000 x 100      =  2.02 TB/day  (both streams)
90-day window   2.02 x 90                 =  181.8 TB
with RF 3       181.8 x 3                 =  545.4 TB

The 90 days is not arbitrary: it is the window Scale and cost already pays for, and the reconciliation job in Reconciliation against billing and who wins reads the same log rather than a second copy.

Data model: pick the grid before you pick the engine

Decide what a stored row is before deciding what stores it. The shape of the data settles two arguments — one row per cell rather than one row per event, and integers rather than floating-point numbers for money — and no database choice can rescue you from getting either one wrong.

How many cells, and how full is each one

The aggregate is a grid of cells. One cell is one combination of the things you group by: this advertisement, in this country, during this one minute.

Stated dimensions: 1,000,000 ads active per day, each live in 3 countries for an average of 480 minutes. Multiply those three to get the number of cells, then divide the day’s events across them to find out how many events land in an average occupied cell.

cells/day       1,000,000 x 480 x 3       =  1,440,000,000
impressions per occupied cell
                20,000,000,000 / 1,440,000,000  =  13.9
clicks per occupied cell
                   200,000,000 / 1,440,000,000  =  0.139

Read those two numbers together and the schema decides itself.

First a term. Granularity, or grain, is how finely time is sliced — minute, hour, or day.

At minute grain, 13.9 impressions collapse into each stored cell, so storing impressions as cells compresses them 13.9x. Clicks do not compress at all. There are 0.139 clicks per cell, which is roughly seven cells per click, so a click-only table at minute grain would store more rows than it has events — the opposite of compression.

The resolution is one row per cell carrying both counters. The click count rides along in the impression row for free, and you never pay for a click-only table.

Here is that row, field by field:

ad_id            8
minute_bucket    4
country          2
impressions      8
clicks           8
spend_micros     8
                ---
                 38     -> 50 B stored, with the sort key and per-row overhead

Tiering: fine grain for days, coarse grain for years

At 50 B per stored row, how much serving state is that? It depends entirely on how long you keep each grain. Three independent stacks — minute rows kept 7 days, hour rows kept 90 days, day rows kept 730 days (two years) — and the coarse tiers turn out to be nearly free.

The cell count changes with the grain because the “480 minutes” in the grid becomes 8 hours or 1 day: 1,000,000 ads x 8 hours x 3 countries at hour grain, 1,000,000 x 1 x 3 at day grain.

minute grain, 7 days     1,440,000,000 x 50   = 72,000,000,000  = 72 GB/day
                         72 x 7                                  = 504 GB
hour grain               1,000,000 x 8 x 3    =     24,000,000  cells/day
                         24,000,000 x 50       =  1,200,000,000  = 1.2 GB/day
                         1.2 x 90                                = 108 GB
day grain                1,000,000 x 1 x 3    =      3,000,000  cells/day
                         3,000,000 x 50        =    150,000,000  = 0.15 GB/day
                         0.15 x 730                              = 109.5 GB
total live               504 + 108 + 109.5                       = 721.5 GB
with RF 3                721.5 x 3                               = 2,164.5 GB

545 TB of raw log against 2.16 TB of serving state — 252x. Two techniques produce that ratio, and it is worth naming both:

• A rollup precomputes coarser summaries from finer ones, so an hourly number is stored rather than summed from sixty minute-rows on demand. That is what makes the query tier a single well-provisioned cluster instead of a data lake.
• Tiering keeps fine grain for a short time and coarse grain for a long time. It is why minute grain is affordable at all: you only keep 7 days of it.

Money is an integer, and the reason is not precision

Money is stored as integer micro-dollars — whole numbers of millionths of a dollar — never as a floating-point number.

The obvious reason would be precision, so dispose of that first. Floating-point addition rounds at every step, and the worst-case relative error of naively summing n numbers is about n x eps, where eps is machine epsilon: the smallest relative gap between two representable double-precision values, about 1.1e-16. Substitute a full day of clicks:

n = 200,000,000 clicks, eps = 1.1e-16
200,000,000 x 1.1e-16                              =  0.000000022

That is 2.2e-8 relative error, against an accuracy budget of 1e-3. Precision is genuinely a non-issue — it has four orders of magnitude of room.

The real reason is determinism: the same inputs must produce the same output bit for bit, every time.

The batch recount adds the same numbers in a different order from the stream, and floating-point addition is not associative — (a + b) + c and a + (b + c) can differ in the last bits. So the reconciler in Reconciliation against billing and who wins would see the two paths disagree in the low bits on every single advertiser, every single day, for no reason at all. A reconciliation that fires on floating-point summation order is a reconciliation you will turn off within a week, and then it protects nothing.

Integer addition is associative. The two summation orders produce bit-identical totals, so any diff the reconciler reports is a real one.

API sketch

Five methods, and two details carry the design: query takes an explicit grain, and billing_snapshot is a separate method rather than a flag on query.

from typing import Iterable, Literal, Protocol


class ClickAggregation(Protocol):
    def ingest(self, events: Iterable[dict]) -> int:
        """At-least-once. Returns accepted count. Duplicates are the
        pipeline's problem, not the collector's -- see section 2.3."""

    def query(self, entity_type: Literal["ad", "campaign", "advertiser"],
              entity_id: int, t0: int, t1: int,
              grain: Literal["minute", "hour", "day"],
              dims: tuple[str, ...] = ()) -> list[dict]: ...

    def top_k(self, advertiser_id: int, t0: int, t1: int, k: int,
              by: Literal["clicks", "spend"] = "spend") -> list[dict]: ...

    def restatements(self, entity_id: int, since: int) -> list[dict]:
        """Every correction applied to a window after it first closed:
        (window, old_value, new_value, reason, applied_at)."""

    def billing_snapshot(self, advertiser_id: int, day: int) -> dict:
        """Immutable, produced only after the batch recount agrees."""

One convention and two design points.

**The convention: `Deep dive.

flowchart TB
    C(["edge collectors<br/>global, 233k events/s avg<br/>701k at peak x3"]) --> L["append-only log<br/>partitioned, RF 3<br/>see ch 20"]
    L --> D["dedup + enrich<br/>event_id set, 24 h TTL"]
    D --> A["windowed aggregator<br/>event-time tumbling 1 min<br/>watermark 30 s"]
    A -->|"absolute cell values"| S[("OLAP store<br/>upsert on cell key<br/>2.16 TB")]
    A --> AM["late / IVT amendments<br/>restatement stream"]
    AM --> S
    L --> AR[("raw archive<br/>object store, 90 d, 545 TB")]
    AR --> B["daily batch recount<br/>T+26 h"]
    B --> R{"reconciler<br/>diff vs stream"}
    S --> R
    R -->|"within threshold"| BI["billing snapshot<br/>immutable"]
    R -->|"outside threshold"| H["hold invoice<br/>page the on-call"]
    S --> Q["query tier<br/>+ 1 s TTL cache"]

    style L fill:#1d3557,color:#fff
    style AR fill:#1d3557,color:#fff
    style BI fill:#1d3557,color:#fff
    style A fill:#495057,color:#fff
    style R fill:#bc6c25,color:#fff
    style H fill:#bc6c25,color:#fff

Reading the colours

Colours follow ch 01’s key.

Blue is an authoritative copy of the data. There are three here, and that is deliberate: the log, the archive that is the log’s long-lived form, and the billing snapshot that is the only number anybody may invoice.

Orange is a box whose timing is forced by something other than processor speed — the reconciler by the batch job’s T+26 h schedule, the invoice hold by a human being available to look.

Grey is the plane that watches and transforms without owning anything. The aggregator’s cells can be deleted and rebuilt from the log at any time.

Nothing here is red, and that is the point of Reconciliation against billing and who wins: every step in this diagram is undoable. A held invoice gets released. A published cell gets restated. The one action that would not be undoable — an invoice mailed on a wrong number — is exactly what the hold exists to prevent.

Follow one click through the pipeline

Ingest. Edge collectors accept the event and write it to an append-only log, partitioned by key and replicated three ways. An append-only log is a file that is only ever added to, never edited, which is what makes replaying it reproducible (ch 20 is that structure’s own chapter).

Dedup and enrich. This stage keeps an event_id set — a table of the unique event identifiers it has already seen — with a 24-hour TTL, meaning time to live, the age at which an entry is discarded. Any event whose identifier is already in the set is dropped.

Aggregate. Surviving events reach the windowed aggregator, which buckets them by event time into 1-minute tumbling windows and closes each window 30 seconds after the fact.

Publish. The aggregator writes absolute cell values into an OLAP store — OLAP is online analytical processing, a database built for aggregating over huge numbers of rows rather than for updating single ones. Each write is an upsert: a write that inserts the row if it is missing and overwrites it if it is present, keyed by the cell.

Amend. Events that arrive too late for their window are neither dropped nor held in memory. They take the second path out of the aggregator into the restatement stream, which recomputes the affected cells, upserts the corrected absolute values, and records what changed and why. Invalid-traffic clawbacks travel the same path.

Audit. The bottom half of the diagram is the second computation. The raw archive keeps 90 days of the original log in object storage, and a batch job recounts a whole day from it at T+26 h. The reconciler compares that recount against what the streaming path stored. Inside threshold, the day is frozen into an immutable billing snapshot and invoiced. Outside threshold, the run stops at hold invoice and pages the on-call engineer — a wrong invoice costs more than a late one.

The log is the system of record, and everything downstream of it is a materialization — a derived copy — that can be deleted and rebuilt from the log at any time. That single property is what makes the recount possible at all, and it is why the archive, not the OLAP store, is the thing you protect.

2. Deep dive

Six mechanisms, one per row of the opening table, each priced in dollars or bytes — and each price rules something out.

1. Windowing, and why event time is not a preference

Windowing sounds like a matter of taste — pick a shape, pick a timestamp, move on. Both choices are forced here: only one of the three window shapes may touch a billable number, and the timestamp that decides a window must be the one written on the device rather than the one read off the server’s clock.

Three window shapes

• A tumbling window is a fixed, non-overlapping bucket. Minute 10:03 holds every event whose timestamp falls in that minute, and every event belongs to exactly one window.
• A sliding window overlaps. A 5-minute window advancing every minute means each event belongs to five windows at once.
• A session window has no fixed boundaries at all. It groups one user’s activity until a gap of inactivity ends it, so its width depends on the data.

Only the first is the default here, and the cost each shape imposes at this traffic is why:

The sliding cost is worth deriving, because candidates reach for sliding windows reflexively. Each event landing in 5 windows means 5 stored cells instead of 1, so multiply the cell count and the storage by 5:

tumbling cells/day                        =  1,440,000,000
sliding 5 min / 1 min step, per event     =  5 windows
sliding cells/day   1,440,000,000 x 5     =  7,200,000,000
storage             7,200,000,000 x 50    =  360,000,000,000   = 360 GB/day

Five times the write amplification — five bytes written for every byte the question needs — and five times the storage, to answer a question you can answer with five reads.

So store tumbling minutes and sum five adjacent cells at query time. That trade moves the cost from the write path to the read path. Read fan-out is the number of stored rows one query has to touch; answering “last 5 minutes” from tumbling minutes multiplies it by 5, and costs nothing at all on the write side.

Read fan-out is the cheap resource here. The query tier serves 2,000 queries/s (Back of envelope what a counting error costs) against 6,945 clicks/s at peak, so a 5x read multiplier lands on the smaller of the two rates.

State that comparison precisely, because it is easy to overclaim. The full event stream at peak is 701,389/s — 350x the read rate. Nothing here says reads outnumber that. The claim is only that reads are small compared to the billable click stream.

Session windows never touch the billing path at all. A session boundary is a heuristic, and you cannot invoice a heuristic.

Event time versus processing time

This is the decision that everything else depends on. Recall the difference: event time is the instant stamped on the device where the click happened; processing time is whenever the pipeline happened to see it.

Processing-time windows are trivially implementable — no watermark, no late data, no state beyond the window currently open. They are disqualified by one property, in the second row of this table:

Reconciliation is a comparison between two computations of the same quantity, so the quantity has to be well-defined independently of which pipeline computed it.

Processing time breaks exactly that. The batch recount reads yesterday’s log at its own speed; the stream job read it live, at whatever speed the network allowed. Under processing-time windows those two runs bucket the same event differently, so they are answering different questions. Their disagreement then carries no information, and the whole reconciliation apparatus in Reconciliation against billing and who wins is theatre.

Event time is not a quality bar. It is the precondition for having a quality bar.

What event time costs: state

The cost of event time is state — the data the aggregator must hold in memory between reading an event and publishing a result.

The reason is mechanical: a window cannot be released until you stop expecting events for it, so every window still open is state you are carrying. The longer you tolerate late events, the more windows stay open at once. Compare tolerating 30 seconds of lateness against tolerating 24 hours:

cells per minute        1,440,000,000 / 1,440       =  1,000,000
state at 30 s lateness (2 minutes of cells in flight)
                        1,000,000 x 2 x 50          =  100,000,000     = 100 MB
state at 24 h lateness
                        1,440,000,000 x 50          =  72,000,000,000  = 72 GB

100 MB fits in a process’s ordinary memory. 72 GB does not.

72 GB means an on-disk state backend, and an on-disk state backend brings three more problems with it: checkpointing — periodically writing the whole aggregator state somewhere durable so a crash does not lose it — plus the recovery time to reload that state, plus a compaction story to stop the on-disk files growing without bound.

That factor of 720 between the two lines is the entire argument for Lambda kappa and the honest pick: you do not buy 24-hour tolerance for late events by widening the stream job’s window.

2. The watermark, priced

The aggregator above closes a window 30 seconds after the fact. That number is a choice, both sides of it can be priced in dollars per day, and the pricing reaches the result the whole chapter turns on: no single value is good enough, which is why the design needs a second read horizon rather than a better-tuned first one.

What a watermark is

A watermark is the pipeline’s promise that it has now seen everything with an event time earlier than T. Once the watermark passes T, every window that ends before T may be closed and published.

It is a guess. The only honest way to make a guess is to price both sides of being wrong:

• Set it too early and you close windows before the stragglers arrive, so you throw away events that were still coming.
• Set it too late and every downstream decision is made on stale numbers.

Below, w is the watermark delay in seconds — how long after a window ends you wait before closing it.

The measured input: event-time skew

Everything in this subsection rests on one measured input. Name it as an input rather than letting it hide inside a table.

Event-time skew is the delay between when an event happened and when the pipeline can see it. It is the sum of network transit, retries, and — dominantly — mobile applications that buffer events while the device is offline and flush them on reconnect. Those are SDKs, software development kits: the libraries an app embeds to report events.

The distribution of that skew is what you choose the watermark against, so every dollar figure below is only as good as this curve.

The distribution is cumulative: by 30 seconds after the event, 98.5% of that minute’s events have arrived. The remaining 1.5% is the tail, and the tail is slow — a full 24 hours only gets you to 99.999%, leaving 0.001% still outstanding. This curve was measured over a week.

delay d      share of events already received
   5 s          92.000 %
  30 s          98.500 %
   2 min        99.600 %
   5 min        99.800 %
  10 min        99.900 %
   1 h          99.980 %
  24 h          99.999 %

Pricing both sides of the choice

Side A: what you drop. Close a window at watermark w and discard everything later, and you lose (1 - share(w)) of the day’s revenue. At w = 30 s, share = 0.985, so you lose 1.5% of $90M.

Side B: what waiting costs. Budget pacing is the control loop that stops showing an advertiser’s ads once they have spent their daily budget. It reads the aggregate to decide.

If the aggregate is w seconds stale, a campaign that has already hit its daily cap keeps being served for w more seconds. That overdelivery is not billable — the platform showed the ads and cannot charge for them. Waiting longer is therefore not free even when nothing is dropped.

To put a rate on side B you need to know how much spend is capped at all. Stated: 30% of spend comes from campaigns that exhaust their daily budget. Convert that to dollars per second of staleness:

capped-campaign spend/day    0.30 x 90,000,000     =  27,000,000    $
their DAY-AVERAGE spend rate 27,000,000 / 86,400   =         312.50 $/s

Each capping campaign overshoots by its own spend rate times w, and summing over all of them gives w x 312.50 dollars of unbillable delivery per day.

State the assumption buried inside that $312.50. It is a day-average rate — total capped spend divided by all 86,400 seconds in the day. That is right only if a capped campaign spends at the average rate right up to the moment it caps. In reality a campaign that exhausts its budget by midday was spending at twice the average; one that caps after six hours, at four times it.

So $312.50/s is a lower bound, and every total below is a floor. The sensitivity check further down re-runs the whole curve at 2x and 4x.

Now both sides together, in dollars per day. Each group is: revenue dropped by closing at w, plus overdelivery caused by being w stale, plus their total.

w =   5 s    dropped   0.08000 x 90,000,000  =  7,200,000
             overshoot       5 x 312.50      =      1,563
             total    7,200,000 + 1,563      =  7,201,563
w =  30 s    dropped   0.01500 x 90,000,000  =  1,350,000
             overshoot      30 x 312.50      =      9,375
             total    1,350,000 + 9,375      =  1,359,375
w =   2 min  dropped   0.00400 x 90,000,000  =    360,000
             overshoot     120 x 312.50      =     37,500
             total      360,000 + 37,500     =    397,500
w =   5 min  dropped   0.00200 x 90,000,000  =    180,000
             overshoot     300 x 312.50      =     93,750
             total      180,000 + 93,750     =    273,750
w =  10 min  dropped   0.00100 x 90,000,000  =     90,000
             overshoot     600 x 312.50      =    187,500
             total       90,000 + 187,500    =    277,500
w =   1 h    dropped   0.00020 x 90,000,000  =     18,000
             overshoot    3600 x 312.50      =  1,125,000
             total       18,000 + 1,125,000  =  1,143,000

Both terms are costs, so the total is what to minimise, and the curve it traces is U-shaped. Dropping dominates at small w — at 5 seconds you are throwing away 8% of the day’s revenue. Overshoot dominates at large w — at an hour you are giving away $1.1M of unbillable delivery. At the day-average rate the minimum sits near 5 minutes, at $273,750/day.

That is the best a single watermark can do. Check it against the accuracy budget from Requirements by expressing it as a share of daily revenue:

best single watermark   273,750 / 90,000,000       =  0.00304
accuracy budget                                    =  0.00100

The best possible single watermark is 0.30% — at least three times over budget.

Now the sensitivity check promised above. Re-run the whole curve at 2x and 4x the day-average capping rate, because the U-curve’s minimum moves when side B gets more expensive:

R = $  312.50/s   optimum w = 300 s   total $273,750/day  =  0.30 % error   [day average]
R = $  625.00/s   optimum w = 300 s   total $367,500/day  =  0.41 % error   [caps at midday]
R = $1,250.00/s   optimum w = 120 s   total $510,000/day  =  0.57 % error   [caps at 6 h]

The conclusion survives every row; the two headline figures do not.

What survives: no single watermark reaches 0.1% under any of the three rates. That is the result the rest of the chapter is built on.

What does not survive: “the minimum sits near 5 minutes” holds only at the day-average rate — at 4x it moves to 2 minutes. And “0.30%” is a floor, not an estimate, because $312.50/s was a lower bound. So quote the error as at least 3x over budget, never as exactly 3x.

This is the load-bearing result of the chapter. Say it in an interview like this: “I priced both sides of the watermark and the optimum still misses the accuracy target by at least 3x, so a single-horizon design cannot work and I need restatement.”

The code below is the same curve as an executable model. watermark_cost is the two-term sum you just read; the assertions pin every figure in the two blocks above, and the loop at the end re-checks the sensitivity rows.

REVENUE_PER_DAY = 90_000_000          # section 1
BUDGET = 0.001                        # the 0.1 % accuracy target
SHARE = {5: 0.92, 30: 0.985, 120: 0.996, 300: 0.998, 600: 0.999, 3600: 0.9998}
CAPPED_RATE = 0.30 * REVENUE_PER_DAY / 86_400        # $/s, the DAY AVERAGE


def watermark_cost(w, rate=CAPPED_RATE):
    """Dollars/day at watermark `w`: stragglers dropped + unbillable overdelivery.
    Both terms are costs, so the total is the thing to minimise."""
    return (1 - SHARE[w]) * REVENUE_PER_DAY + w * rate


assert round(CAPPED_RATE, 2) == 312.50
assert round(watermark_cost(30)) == 1_359_375
assert round(watermark_cost(120)) == 397_500
assert round(watermark_cost(300)) == 273_750
assert round(watermark_cost(600)) == 277_500
assert round(watermark_cost(3600)) == 1_143_000

best = min(SHARE, key=watermark_cost)
assert best == 300, best
assert abs(watermark_cost(best) / REVENUE_PER_DAY - 0.00304) < 1e-5
assert watermark_cost(best) / REVENUE_PER_DAY > BUDGET      # over budget, and

# ... by AT LEAST 3x. $312.50/s is the day-average spend rate of capped
# campaigns, so it is a lower bound: a campaign that caps at midday spent at 2x
# it, one that caps at 6 h at 4x. Re-run the curve at both. The optimum moves
# and the error grows; what does not move is that no single watermark qualifies.
SENSITIVITY = {312.50: (300, 273_750), 625.00: (300, 367_500),
               1250.00: (120, 510_000)}
for rate, (w_star, total) in SENSITIVITY.items():
    got = min(SHARE, key=lambda w: watermark_cost(w, rate))
    assert got == w_star, (rate, got, w_star)
    assert round(watermark_cost(got, rate)) == total
    assert watermark_cost(got, rate) / REVENUE_PER_DAY > 3 * BUDGET

Name the trap before the interviewer does: a perfect watermark is impossible. You cannot distinguish “no events yet” from “no events ever” without a bound on delay, and a phone in airplane mode gives you no bound. Watermarks are heuristics with a cost function attached, which is why you price them instead of arguing about them.

Say the dependency out loud too, because it is the assumption the rest of the chapter stands on: every number in this subsection is a function of the event-time skew distribution above, and nothing else.

Shorten the tail and a single watermark becomes sufficient, which deletes the amendment path entirely. Lengthen it and the billing date itself has to move. The assumption ledger records this as the chapter’s load-bearing assumption for exactly that reason.

3. Exactly-once, honestly

Late events were half the accuracy story. The other half is duplicates: delivery is at-least-once, so the same click can arrive twice, and a pipeline that bills on it has to make the second copy change nothing.

Exactly-once delivery does not exist

End-to-end exactly-once does not exist as a delivery property. Ch 20 covers why; the argument is one sentence long.

A sender that gets no acknowledgement cannot tell a message that was lost from an acknowledgement that was lost. So it must either resend, which risks a duplicate — that is at-least-once — or not resend, which risks a loss — that is at-most-once. There is no third option across an unreliable link.

What does exist is at-least-once delivery plus an effectively-once effect: duplicates arrive, and applying one twice changes nothing. There are exactly two ways to build that effect, and the rest of this subsection prices both.

Design A: idempotent write, keyed on event_id

Idempotent means applying an operation twice leaves the same result as applying it once.

The collector stamps each event with a UUID at the edge — a universally unique identifier, a 128-bit random value with a negligible chance of collision (ch 07 if you want it sortable and smaller).

Everything downstream keeps a set of the identifiers it has already seen within a dedup window — the time span over which duplicates are considered possible, 24 hours here — and silently drops repeats. Size that set for the click stream, then for both streams together:

click ids in a 24 h window                          =    200,000,000
open-addressed set, 16 B key + 8 B slot metadata, load factor 0.5
   200,000,000 x 24 / 0.5                           =  9,600,000,000   = 9.6 GB
same for both streams
   20,200,000,000 x 24 / 0.5                        = 969,600,000,000  = 970 GB

Two constants in that block need naming. An open-addressed set is a hash table that stores its entries directly in one flat array rather than in linked lists. Its load factor is how full that array is allowed to get — 0.5 here, meaning the array is twice the size of the data it holds, which is where the / 0.5 comes from. The x 24 is 24 bytes per entry: a 16-byte key plus 8 bytes of slot metadata.

9.6 GB fits in memory on a single machine with room to spare. 970 GB does not.

So deduplicate the money stream exactly, and do something cheaper for impressions.

The cheaper thing is a rotating ring of 24 hourly Bloom filters. A Bloom filter is a compact bit array that answers “have I seen this?” using a few hash functions. It never says no about something it has actually seen, but it will occasionally say yes about something it has not — a false positive. The rate of those, written p, is what you trade against size: smaller p costs more bits per element.

Size it at p = 1e-6, one wrong answer per million. The first line is the standard optimal-sizing formula for a Bloom filter, -ln(p) / (ln 2)^2 bits per element:

bits per element at p = 1e-6:   -ln(p) / (ln 2)^2
                        13.8155 / 0.4805            =  28.75  bits
bytes per element       28.75 / 8                   =   3.59  B
events per hourly filter
                        20,200,000,000 / 24         =  841,666,667
per filter              841,666,667 x 3.59          =  3,021,583,333   = 3.02 GB
all 24                  3.02 x 24                   =  72.5 GB

72.5 GB against 970 GB, so the filter is 13x cheaper than the exact set for the impression stream.

Now the honest part. A Bloom false positive here means a real event is wrongly identified as a duplicate and discarded — a silent undercount. Price that as if you had used the filter on clicks, where the money is:

clicks lost   200,000,000 x 0.000001                =  200 /day
their value   200 x 0.45                            =   90 $/day
share         90 / 90,000,000                       =  0.000001

$90/day, a thousand times inside the accuracy budget — and you still do not use it on the money stream.

The reason is not the error rate. It is that the auditor’s question is “which 200 clicks?”, and a Bloom filter cannot enumerate what it dropped. It is a bit array; there is nothing in it to list.

So: exact dedup on clicks, probabilistic dedup on impressions. The line that gets you hired is “I chose the exact structure for the stream where I have to be able to name the missing rows, not the stream where the error is larger.”

The second idempotence, which is free

There is a second kind of idempotence available here, and it costs nothing. Emit the cell’s absolute value, not a delta.

Compare the two writes:

INCREMENT clicks BY 1  WHERE cell = (ad, minute, country)     -- not idempotent
UPSERT    clicks =  n  WHERE cell = (ad, minute, country)     -- idempotent

An INCREMENT applied twice adds 2. An UPSERT applied twice still leaves n.

For the upsert to be correct, the aggregator has to compute a window deterministically from a slice of the log it can re-read at will. Given that, replaying the slice writes the same n again. A duplicated output is then harmless — the second write lands on top of the first and changes nothing — and the dedup store only has to defend against duplicated input.

Making the write a SET rather than an ADD moves the exactly-once problem from the sink, where it is expensive, to the source, where it is 9.6 GB.

Design B: transactional sink

The sink is wherever the results are written. In this design the processor commits two things as one indivisible unit: how far it has read from the log, and the writes it produced from that reading. Either both land or neither does, so a crash can never leave the read position ahead of the writes.

Kafka transactions do this, as does a two-phase-commit sink in Flink. Two-phase commit is the protocol where a coordinator first asks every participant to promise it can commit, then tells them all to do it. Flink is a widely used stream-processing engine.

The mechanism is real. It costs three things:

Pick A. The idempotent absolute-value upsert plus a bounded dedup set is cheaper, and it survives a sink that has no transaction support at all.

A third reason decides it: A makes replay — deliberately re-reading and reprocessing a slice of the log — a supported everyday operation rather than a recovery emergency. Lambda kappa and the honest pick needs replay to be routine, so a design that makes replay scary is not usable here.

The code below is Design A in miniature. dedup_and_aggregate folds one batch of events into cells, skipping ids it has already seen; emit writes those cells with upsert, never increment. The ValueError near the top is a guard, not decoration — the paragraph after the block explains the failure it catches.

MICROS_PER_SECOND = 1_000_000       # event_time is epoch MICROseconds


def dedup_and_aggregate(events, seen, cells, window_secs=60):
    """One aggregation step. Emits ABSOLUTE cell values so a replay of the
    same log slice produces byte-identical output (section 2.3).

    seen  -- set of event_id, scoped to THIS window and discarded with it
    cells -- dict[(ad_id, bucket, country)] -> [impressions, clicks, micros]

    The scoping rule is a contract, not a detail. A replay of one log slice
    must reset BOTH `seen` and `cells`: keeping `seen` drops the window
    silently, keeping `cells` doubles it. Both directions are tested below.
    """
    if seen and not cells:
        raise ValueError(
            "`seen` survived into an empty `cells`: every event would be "
            "refused as a duplicate and the window would come back empty. "
            "Reset both, or neither.")
    accepted = 0
    for e in events:
        if e["event_id"] in seen:
            continue
        seen.add(e["event_id"])
        accepted += 1
        bucket = e["event_time"] // (window_secs * MICROS_PER_SECOND)
        key = (e["ad_id"], bucket, e["country"])
        cell = cells.setdefault(key, [0, 0, 0])
        if e["kind"] == "impression":
            cell[0] += 1
        elif e["kind"] == "click":
            cell[1] += 1
            cell[2] += e["price_micros"]        # integer micro-dollars
        else:
            raise ValueError(f"unknown event kind {e['kind']!r}")
    return accepted


def emit(cells, sink):
    """Upsert, never increment. Replaying the same slice is a no-op."""
    for key, (imps, clicks, micros) in cells.items():
        sink.upsert(key, impressions=imps, clicks=clicks, spend_micros=micros)

The dedup set’s scope is a contract, and the chapter has to state it or the crash-recovery row in Failure modes is not true.

seen is scoped to one window and discarded with it. A replay of one log slice must reset both seen and cells. Get that half-right and you fail in one of two directions:

• Keep seen, reset cells → every event is refused as a duplicate and the window comes back empty. Silent undercount.
• Keep cells, reset seen → every event is added on top of numbers that already include it and the window doubles. Silent overcount.

Both directions are worth testing, because neither is loud on its own. The block below runs the aggregator four times: once for the correct result, once to prove a replay is byte-identical, and once in each failure direction.

MINUTE = 60 * MICROS_PER_SECOND
T0 = 1_700_000_040 * MICROS_PER_SECOND         # a minute boundary, in MICROseconds
EVS = [{"event_id": f"e{i}", "ad_id": 7, "country": "US", "kind": "impression",
        "event_time": T0 + i * 6 * MICROS_PER_SECOND,   # 0,6,...,54 s: ONE minute
        "price_micros": 0} for i in range(10)]
EVS.append({"event_id": "c0", "ad_id": 7, "country": "US", "kind": "click",
            "event_time": T0, "price_micros": 450_000})

first: dict = {}
assert dedup_and_aggregate(EVS, set(), first) == 11
assert len(first) == 1, "eleven events inside one real minute are ONE cell"
# 10 impressions and 1 click, NOT 11 impressions: a click is not an impression,
# because the impression that preceded it was sent as its own event.
assert first[(7, T0 // MINUTE, "US")] == [10, 1, 450_000], first

again: dict = {}
dedup_and_aggregate(EVS, set(), again)
assert first == again, "replay of one slice must be byte-identical"

stale = {k: list(v) for k, v in first.items()}
dedup_and_aggregate(EVS, set(), stale)
assert stale != first, "replaying into surviving cells double-counts -- reset them"

try:                       # the other direction: `seen` survives, `cells` does not
    dedup_and_aggregate(EVS, {e["event_id"] for e in EVS}, {})
except ValueError:
    pass
else:
    raise AssertionError("a surviving dedup set must be refused, not silently "
                         "returned as an empty window")

try:
    dedup_and_aggregate([{**EVS[0], "event_id": "x", "kind": "conversion"}], set(), {})
except ValueError:
    pass
else:
    raise AssertionError("an unknown kind must raise, not be counted as an impression")

4. Lambda, kappa, and the honest pick

The watermark priced ended with an unmet budget: the best single watermark misses the 0.1% target by at least 3x. Two textbook architectures claim to close that kind of gap, both fail here in their standard form, and the variant that works gets the error to 0.011%.

Lambda, and the objection that kills it

The classic lambda architecture runs a fast approximate stream path beside a slow exact batch path, and serves the union of the two.

State the real objection, which is not about cost or operations: two serving paths means two implementations of “what counts as a click”, so the difference you measure between them is dominated by drift between your own two codebases rather than by anything about the data. You have built a measuring instrument whose noise floor is its own construction, and it can no longer tell you whether the pipeline is wrong.

Kappa, and why it is not quite enough

The kappa architecture is the reply: one code path, streaming only. If you need a number recomputed, you replay the log through that same code.

Its claim is that one exact stream path is enough. At these numbers it is not, because The watermark priced showed no single watermark reaches 0.1%.

The fix: one pipeline, two read horizons

The resolution is not a second serving path. It is one pipeline read at two different horizons — the same stored cells, consulted at two different moments, with two different guarantees attached:

Events that arrive after w are neither dropped nor held in the aggregator’s state — holding them was the 72 GB from Windowing and why event time is not a preference.

They go instead to an amendment stream: a separate, low-volume path that re-reads the cells the late event belongs to, recomputes their absolute values including it, and upserts the corrected numbers together with a restatement record saying what changed and why.

Price the result. Two costs survive this design — pacing is still 30 s stale, and a sliver of events is still missing at the T+24 h cut — so add them and compare against the budget: 0.1% of $90M is $90,000/day.

overshoot at w = 30 s     30 x 312.50                =   9,375 $/day
residual dropped after T+24 h
                          0.00001 x 90,000,000       =     900 $/day
total                     9,375 + 900                =  10,275 $/day
as a share of revenue     10,275 / 90,000,000        =  0.000114
budget headroom           0.001 / 0.000114           =  8.77

0.011% against a 0.1% budget — 8.8x of margin, where the best single watermark was at least 3x over.

That is the whole payoff of the chapter, and it comes from closing early rather than late. Waiting less costs almost nothing because the amendment path recovers what the early close missed.

Now size that amendment path, so you know what you have signed up for. Its volume is the tail beyond 30 s — 1.5% of events, from the skew table in The watermark priced:

amendment events/day      0.015 x 20,200,000,000     =  303,000,000
amendment events/s        303,000,000 / 86,400       =  3,507

3,507/s against 233,796/s on the main path is 1.5% of the ingest rate — one event for every 67 the main path already handles. That is why the amendment path can afford to be simple, sequential, and heavily instrumented.

The block below pins those figures. It recomputes the error total, the 8.8x headroom, and the amendment rate, and asserts that the two-horizon design is inside budget while the best single watermark is not.

REVENUE_PER_DAY = 90_000_000                       # section 1
BUDGET = 0.001                                     # the 0.1 % accuracy target
CAPPED_RATE = 0.30 * REVENUE_PER_DAY / 86_400      # $312.50/s, section 2.2

overshoot = 30 * CAPPED_RATE                       # pacing stale by w = 30 s
residual = (1 - 0.99999) * REVENUE_PER_DAY         # still missing at T+24 h
assert round(overshoot) == 9_375
assert round(residual) == 900
assert round(overshoot + residual) == 10_275

error = (overshoot + residual) / REVENUE_PER_DAY
assert abs(error - 0.000114) < 1e-6
assert round(BUDGET / round(error, 6), 2) == 8.77
assert error < BUDGET, "the two-horizon design is inside budget"
assert 273_750 / REVENUE_PER_DAY > BUDGET, "the best single watermark is not"

AMENDMENTS_PER_DAY = 0.015 * 20_200_000_000        # the tail beyond 30 s
assert round(AMENDMENTS_PER_DAY) == 303_000_000
assert round(AMENDMENTS_PER_DAY / 86_400) == 3_507
assert round(AMENDMENTS_PER_DAY / 20_200_000_000 * 100, 1) == 1.5

The pick, stated the way you should say it: “Kappa with restatement, plus a batch recount that is an audit rather than a serving path. One definition of a click, one code path that produces billable numbers, and a second computation whose only job is to disagree with the first.” The batch job in Reconciliation against billing and who wins is not lambda’s batch serving layer, because nobody ever queries it; it exists only to be compared against.

What this choice costs

State the costs before the interviewer asks. All three come from the same source: a published number is now allowed to change after publication.

5. Hot partitions: salt, then merge

Traffic this concentrated breaks any partitioning scheme that treats keys as interchangeable. The skew — how much more traffic the busiest partition gets than the average one — comes straight out of the traffic model, and the standard answer to uneven load does nothing about it.

Why consistent hashing does not help

One advertiser is 10% of all clicks. The reflex answer is consistent hashing, and it is wrong here.

Consistent hashing places both keys and servers on a circle by their hash values and gives each key to the next server clockwise. Each server appears at many points on the circle — its virtual nodes — so adding or removing a server moves only a small share of keys.

What consistent hashing does not fix derives why that does not help with a hot key. The ring balances the keyspace. A single key has exactly one hash, lands in exactly one arc of the circle, and belongs to exactly one owner, no matter how many virtual nodes you configure.

Here is the same argument in this system’s units, with P = 64 log partitions keyed by advertiser_id. The hot advertiser’s partition carries its 10% plus its fair share of everything else:

uniform share per partition    1 / 64                =  0.015625
hot advertiser's share                               =  0.10
its partition's share          0.10 + 0.90 / 64      =  0.1140625
skew over the mean             0.1140625 / 0.015625  =  7.30

7.30x on one partition, and no number of virtual nodes moves it, because the hot thing is the aggregation key itself, not where that key is placed.

That number is expensive because you must provision every machine in the fleet for whatever the busiest one does. 7.30x skew means paying for 7.3x the hardware to do 1x the work.

The fix: salt the hot key

The fix is to stop the hot key being a single key. Salt it — attach a small extra component to the key so one logical key becomes S physical ones, spread across S partitions.

With S = 16, the hot advertiser’s 10% is split into sixteen pieces of 0.625% each, and only one of those pieces lands on the busiest partition:

salt fanout S = 16, salt = stable_hash(event_id) mod 16
hot key's share per salt       0.10 / 16             =  0.00625
hottest partition now          0.00625 + 0.90 / 64   =  0.0203125
skew                           0.0203125 / 0.015625  =  1.30

7.30x -> 1.30x, for a fleet that is now nearly balanced.

The salt is derived from event_id, not from a counter or a random number, and that detail is the whole reason salting is safe here. The same event always produces the same salt, so a replay routes it to the same partition and the pipeline stays deterministic — which Windowing and why event time is not a preference established is non-negotiable.

“Deterministic” has to mean across processes, and that rules out the language’s own hash function. Python’s hash() on a string is randomised per interpreter, so the same event_id salts differently after a restart. Every hot key re-partitions, and the batch recount then disagrees with the stream by construction — which is exactly the failure Reconciliation against billing and who wins says the reconciler can never diagnose, because the diff looks like ordinary data loss. Use a named, stable digest and pin its output with a test.

The cost: a second merge stage

A salted key produces S partial aggregates per window, and something has to add them back together.

Size that merge, because “add a merge stage” sounds expensive and is not. The arithmetic is a chain: only keys above the uniform share are worth salting, which caps how many advertisers qualify; each of those owns some ads in some countries; each of those cells is split S ways.

salt only keys above the uniform share    1 / 64      =  0.015625
so at most                               1 / 0.015625 =  64 advertisers qualify
ads per advertiser        1,000,000 / 100,000         =  10
cells per hot advertiser  10 ads x 3 countries        =  30
partial rows per minute   64 x 30 x 16                =  30,720
merge input rate          30,720 / 60                 =  512 rows/s

Five hundred and twelve rows per second.

The unit matters, so state it. A cell is (ad, minute, country), not (advertiser, minute). A hot advertiser owns roughly ten ads live in three countries, so the merge sees 30 rows per salt per minute, not one. Getting that wrong understates the merge by 30x.

Compare the two policies. Salting everything would multiply the entire 1.44e9-cell grid by 16. Salting only the provably-hot keys costs a merge stage a laptop could run.

Two pieces of machinery make “provably hot” work:

• A rolling heavy-hitters sketch on the ingest path detects which keys are hot. It is a small fixed-size structure that tracks approximately which keys are most frequent, without storing a counter per key.
• The resulting salt map — the list of which keys are currently split — is published alongside the partition assignment, so the producing and consuming sides always agree on which keys are salted. A stale salt map is a real failure mode; Interviewer pushback walks through what it looks like.

The code below is the salt and the merge. Note _stable_hash: the docstring explains why it is not Python’s hash(), and the pinned tuple of salts at the bottom is what makes a change to that hash fail loudly here instead of silently re-partitioning yesterday’s data.

import hashlib


def _stable_hash(s: str) -> int:
    """NOT Python's hash(): str hashing is randomised per process (PYTHONHASHSEED),
    so a restart would re-route every hot key and the recount could never agree.
    """
    return int.from_bytes(hashlib.blake2b(s.encode(), digest_size=8).digest(), "big")


def salted_key(event, hot_keys, fanout=16):
    """Deterministic salt for provably-hot keys only (section 2.5).

    The salt comes out of event_id through a STABLE hash, so a replayed event
    lands in the same partition in a different process, on a different machine,
    next year -- which is what section 2.1 means by reproducible.
    """
    advertiser = event["advertiser_id"]
    if advertiser not in hot_keys:
        return (advertiser, 0)
    return (advertiser, _stable_hash(event["event_id"]) % fanout)


def merge_partials(partials):
    """Second stage: sum the S partial cells back into one row per window."""
    merged: dict = {}
    for (advertiser, _salt), window, imps, clicks, micros in partials:
        row = merged.setdefault((advertiser, window), [0, 0, 0])
        row[0] += imps
        row[1] += clicks
        row[2] += micros
    return merged


HOT = {42}
IDS = ["evt-%04d" % i for i in range(8)]
SALTS = tuple(salted_key({"advertiser_id": 42, "event_id": i}, HOT)[1] for i in IDS)
# Pinned to literals so that changing the hash fails HERE rather than silently
# re-partitioning yesterday's data. With Python's own hash() this tuple is a
# different one in every process, which is the bug this block exists to catch.
assert SALTS == (9, 0, 9, 0, 7, 4, 1, 1), SALTS
assert _stable_hash("evt-0000") == 8306334570326659657
assert salted_key({"advertiser_id": 7, "event_id": "evt-0000"}, HOT) == (7, 0)
assert all(0 <= s < 16 for s in SALTS)
assert merge_partials([((42, s), 0, 1, 1, 450_000) for s in range(16)]) == {
    (42, 0): [16, 16, 7_200_000]}

Two other kinds of skew are worth naming, because interviewers conflate them with this one.

A hot cell — one ad, one minute, a sudden spike — is the same problem at a finer key, and exactly the same salt handles it.

A hot query — everyone loading the same advertiser’s dashboard at once — is not a write problem at all. It is solved by the 1-second cache in the architecture diagram, which serves every repeat request from one stored answer until it expires, exactly as What consistent hashing does not fix prescribes for hot reads.

6. Reconciliation against billing, and who wins

Everything so far produced a number accurate enough to bill. Turning it into an invoice is a loop with four moving parts: something that recomputes the day, two alert thresholds that decide “close enough”, a rule for which number wins a disagreement, and one class of failure that escapes all of it.

Reconciliation is the practice of computing the same quantity twice by independent means and comparing the answers.

The streaming aggregate is never the billed number. The billed number is whatever a recount from the immutable log says it is, because the log is the artifact an auditor, an advertiser’s own tracking system, and a courtroom can each evaluate for themselves.

The loop

1. At T+26 h — two hours past the last meaningful amendment — a batch job reads the raw archive for day T and recomputes every advertiser’s clicks and spend from scratch. Different code, same definition, same event-time boundaries.
2. The reconciler diffs batch against stream, per advertiser and in aggregate.
3. Inside threshold: the streaming table is frozen into an immutable billing_snapshot and invoiced.
4. Outside threshold: the invoice is held, not corrected automatically, and a human looks.

Two thresholds, and why they differ by 5x

The reconciler needs a number that separates “close enough to bill” from “stop and look”. It uses two, at different levels:

per-advertiser daily alert                          =  0.0005     (0.05 %)
aggregate daily alert                               =  0.0001     (0.01 %)

The aggregate threshold is five times tighter than the per-advertiser one. That looks backwards at first — the aggregate covers more data, so surely it should be looser — and it is deliberate.

The reason is how independent random errors combine. They partly cancel when you add them up: summing n of them grows the total absolute error by sqrt(n), not by n. Divide by a total that grew by n, and the relative error of the total shrinks by sqrt(n).

So if per-advertiser discrepancies really were independent noise across 100,000 advertisers, the aggregate discrepancy you would expect is:

0.0005 / sqrt(100,000)   =  0.0005 / 316.23         =  0.00000158
against the aggregate alert   0.0001 / 0.00000158   =  63.2

0.00016% — sixty-three times below the aggregate alert.

So an aggregate discrepancy of 0.01% is mathematically incompatible with independent per-advertiser noise. If the aggregate alert fires, the cause can only be a systematic error — one that pushes every advertiser the same direction at once. A dropped partition. A timezone boundary. A schema change.

The two thresholds are therefore looking for two different failures: the per-advertiser one for localized bugs, the aggregate one for systematic ones. This arithmetic is what lets them be told apart.

Who wins a disagreement

When the two numbers disagree, the batch wins. Three rules follow:

• Bill the lower of the two, then true up. An underbill is a correction you add to next month’s invoice; an overbill is a refund, a credit memo, and a trust problem. The asymmetry priced in Back of envelope what a counting error costs decides the tiebreak.
• Never silently overwrite. Every correction lands in restatements with a reason code. “The number changed and nobody can say why” is the failure that costs accreditation.
• Investigate the diff, not the total. A per-advertiser diff distribution tells you which failure you have: a few advertisers wrong by a lot is a hot-key or salt-map bug; everyone wrong by the same small factor is a boundary or a dedup-window bug.

What neither number can catch

Both computations read the same log, so any bug upstream of the log makes both of them wrong in exactly the same way. A collector that never sent the event is invisible to both, and their diff is zero.

That is why the ingest path is instrumented separately from the pipeline. The edge collector emits a per-minute count of what it sent, and the pipeline compares that count against what it received. The only defense against a shared-mode failure is an independent counter as close to the source as you can put one.

Invalid traffic makes restatement mandatory

IVT classification is retroactive. A click looks fine when it arrives and is provably a bot six hours later, once the fraud system has seen enough of that user’s behaviour.

Removing it is a clawback: a downward amendment to a cell that has already been published. A pipeline with no amendment path has only two options, and both are bad — bill the advertiser for fraud, or delay every invoice until the IVT window has closed. That is why the amendment path is a functional requirement here and not an optimization.

3. Bottlenecks and scaling

What binds first at each scale is a different question from what is expensive, and it is the one this table answers.

Three terms in it need naming first:

• Shuffle — the network step that moves every event with the same key onto the same machine so it can be aggregated there. It is usually the dominant cost in a streaming job.
• RTT — round-trip time, how long a packet takes to reach another region and come back.
• Single-flight — the rule that when many identical requests arrive while one is already in progress, exactly one of them does the work and every waiter gets that same answer.

The rows run top to bottom in increasing scale, except for the last three, which apply at any rate.

The multi-region case is worth one more sentence because it is a common follow-up. Regional partial aggregates plus a global merge is the only shape that does not put a 150 ms round trip in the ingest path, and it composes with Hot partitions salt then merge for free: a region is just a salt whose value you did not get to choose.

4. Failure modes

These are the things that actually go wrong, what each looks like from the outside, and what you do about it.

One term first. NTP is the Network Time Protocol, the service that keeps machine clocks agreed with each other. Every event-time system depends on it, because an edge collector with a wrong clock writes events into the wrong minute and nothing downstream can detect that — the timestamp looks perfectly valid.

5. Alternatives rejected

A rejection with a number attached is worth more than a preference, so every design considered and discarded here comes with its price.

Three terms appear in the table:

• Relative standard deviation (sd) — the typical size of an estimate’s error, expressed as a fraction of the quantity being estimated. A 9.5% relative sd on 1,000 clicks means a typical run is off by about 95 clicks.
• HyperLogLog — a compact sketch that estimates how many distinct items a stream contained, using a few kilobytes instead of a list of them.
• OLTP — online transaction processing, a database tuned for many small reads and writes of individual rows. It is the opposite of the OLAP workload this pipeline serves.

6. Interviewer pushback

The same material, in the form it actually gets asked — each answer stated the way it should be said out loud.

“Your pipeline says exactly-once. Prove it.”

It is not exactly-once delivery — that property does not exist across an unreliable link, for the reason ch 20 gives. It is at-least-once delivery with an effectively-once effect, from two independent mechanisms: a dedup set on event_id that costs 9.6 GB for a 24-hour window on the click stream, and an output that writes absolute cell values rather than increments, so replaying a log slice is a no-op. The second one is the important one, because it means recovery and reconciliation use the same code path as normal operation.

“Why not just set the watermark to an hour and be done?”

Because I priced it. An hour recovers 99.98% of events, worth $18,000/day of otherwise-dropped revenue — but it makes budget pacing an hour stale, and at $312.50/second of capped-campaign spend rate that is $1,125,000/day of unbillable overdelivery. Net $1.14M/day, against $273,750 at the 5-minute optimum. That $312.50 is a day average, so those totals are floors and the optimum moves earlier if capped campaigns spend faster. The whole point of The watermark priced is that even the optimum misses the 0.1% target by at least 3x, which is why I close fast and amend instead of tuning a single number.

“A mobile SDK batches events for six hours. Walk me through one of those events.”

It arrives past the 30-second watermark, so the fast path has already closed and published its cell. It is not dropped and it does not sit in the aggregator’s state — that would be the 72 GB. It goes to the amendment stream, which re-reads the affected cell, recomputes the absolute value including the new event, and upserts it with a restatement record naming the reason. Dashboards see the number change; the invoice, which is not cut until T+24 h, only ever sees the corrected value. The residual is the 0.001% still outstanding after 24 hours: $900/day.

“Your streaming number and your batch number disagree by 0.4% for one advertiser. What do you do?”

Hold that advertiser’s invoice, not the whole run. Then use the shape of the diff: 0.4% on one account against a 0.05% per-account threshold, with the aggregate inside 0.01%, is a localized failure — and the first thing I check is whether that advertiser is on the salt map, because a stale salt map means one of 16 partial aggregates never got merged, which produces exactly a single-account shortfall of roughly 1/16. If the aggregate had also breached, I would be looking for a systematic cause instead, because Reconciliation against billing and who wins shows independent noise cannot produce an aggregate breach.

“What stops you from billing for bot traffic?”

Nothing in the counting pipeline — IVT classification is a separate system, and crucially it is retroactive: a click can be reclassified hours later. That is a second, independent reason the design needs downward restatement, and it is why the amendment path is a functional requirement rather than an optimization. The billing snapshot is taken after the IVT window closes, and the pre-filter count is retained so the clawback is auditable in both directions.

“Where does the ML model fit?”

It does not, and keeping them apart is deliberate. ml/08 predicts a click before the auction; this pipeline counts one after it happened. They meet at exactly one place: this pipeline’s output is that model’s training label, so a counting bug becomes a training-data bug on a 24-hour delay. That is one more reason the restatement record exists — the label pipeline needs to know which rows changed.

The assumption ledger

Every design is a set of assumptions with a diagram attached, and the diagram is only correct relative to them. Everything this chapter has leaned on is collected here, so you can state the design’s foundations in twenty seconds and say what replaces the design when each one fails.

Each assumption goes in one of three bins:

• State it — you are free to pick, and being wrong costs a re-derivation and nothing more.
• Ask it — the answer moves a policy or a threshold, and it is worth an interviewer’s time.
• Load-bearing — if it is wrong the design is not suboptimal, it is invalid. A box appears or disappears, rather than the count inside a box changing.

The one-line test, from ch 03: move the assumption an order of magnitude in each direction and ask whether the set of boxes changes, or only the number of machines inside them.

The table’s last column is the useful one. It says what design you would build instead if the assumption turned out false.

The sentence that makes this visible to an interviewer: “This design rests on four things. One, the measured event-time skew — 98.5% inside 30 seconds but a tail that runs to 24 hours — which is what makes a single watermark impossible and restatement mandatory. Two, that the device’s own clock is good enough to stamp event time, without which reconciliation compares two different questions. Three, that the billed number must be independently recomputable from the raw log, which is what buys the 545 TB archive. Four, that downstream consumers can accept a number being corrected, because if they cannot, there is no real-time dashboard at all.”

Cheat sheet

Every line below is derived somewhere above; this table is the recall test, not the explanation.