How to design an ad-click aggregation pipeline

In this lesson, we’ll build the system that counts one kind of event (somebody clicked an advertisement) and turns those counts into two outputs: a live dashboard number and an invoice. Counting clicks sounds trivial. It stays trivial right up until you remember the number gets multiplied by a price and mailed to a customer, and then every shortcut you would take for an analytics pipeline turns into a refund, an audit, or a lawsuit.

By the end you’ll be able to:

• Say what a window is, and why a window is defined by the timestamp written on the event, not 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.
• Reason about what “exactly-once” can and cannot mean once money is involved.
• Walk 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.

Two different artifacts come out, 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.

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

Why this one cannot be “eventually roughly right”

Most aggregation systems are 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. An analytics pipeline that loses 1% of its events has a rounding error, but a billing pipeline that loses 0.01 x 90,000,000 = 900,000 dollars a day has a lawsuit. The whole design rests on that difference.

Ownership split with the ML lesson

Two lessons touch ad clicks, and they own different halves.

The ad-click-prediction lesson 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 lesson 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 share one input, the traffic model, stated there and cited here.

The argument, in six steps

Six claims carry the argument, each settled by one number.

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:

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, priced in The watermark priced.
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, in increasing order of cost. The first two are ordinary latency problems; only the third is specific to billing.

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 and forward events. 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, window, grain): return counts and spend for an entity over a time range at a chosen grain. The same stored cells are read at two horizons, a fast dashboard read and a final billing read.
• Filter invalid traffic (IVT) (clicks generated by bots or fraud, not by 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. The accuracy target is the unusual one; the next subsection derives it.

Four terms in that table are used through the rest of the lesson:

• 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 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 the same bytes today’s count read.

The shape of that table is worth stating early: ingest availability and billing accuracy are hard while query latency is not, the reverse of most systems.

Back-of-envelope: what a counting error costs

One number justifies the whole design: what a tenth of a percent of miscounting is worth.

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

The traffic model comes from the ad-click-prediction lesson: 2.0e10 (20 billion) ad impressions per day, a click-through rate (CTR) of 1.0% (the fraction of impressions clicked), and an average cost per click (CPC) of $0.45.

That is 200M clicks/day, so about $90M/day of click revenue. A tenth of a percent of it is $90,000/day, or $32.9M/yr: the price of a 0.1% miscount.

The error is asymmetric. 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 costs more than the money.

Concentration makes it worse, because errors do not average out per customer. If the largest advertiser is 10% of spend, that is $9M/day on one account, and 0.1% of it is $9,000/day of exposure on a single customer.

Rates: what the pipeline handles per second

Divide each per-day figure by 86,400 seconds for a per-second average, then multiply by 3 for peak (the busiest hour runs at about three times the daily average, and peak is what you provision for).

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 rate is the one workload figure nothing upstream measures, so build it from a stated assumption: 100,000 advertisers, 2% with a dashboard open during the peak hour, each refreshing once a second (matching the 1-second staleness budget). That is 100,000 x 0.02 = 2,000 dashboards, so 2,000 queries/s.

Say which stream you mean, because the two differ by a hundred times. Reads (2,000/s) are smaller than the billable click stream (6,945/s at peak), but far smaller than the full firehose (701,389 events/s). Reads never outnumber the firehose, and that headroom 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 here, recorded as such in the assumption ledger. Halve or double it and nothing changes shape, because the query tier sits behind a 1-second cache either way.

Event size, bandwidth, and retention

A single click record:

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 with framing and a compact binary encoding.

Ingest bandwidth is 233,796 events/s x 100 B = 23.4 MB/s, and with a replication factor of 3 (the log keeps every event on three separate machines, so losing one loses nothing) it is 70.1 MB/s. That fits on a single machine’s NIC (a network interface card, one gigabit ≈ 125 MB/s), so the whole firehose would fit on one machine. You split it across many 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 traffic is already divided into partitions: independent slices, each handled by its own machine.

Raw retention, both streams: 20.2e9 events/day x 100 B = 2.02 TB/day, over a 90-day window with RF 3 that is 545 TB. The 90 days is the window the ad-click-prediction lesson already pays for, and the reconciliation job reads the same log, not 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 instead of one row per event, and integers instead of floats for money) and no database choice can rescue 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. That is 1e6 x 480 x 3 = 1.44e9 cells/day. Dividing the day’s events across them:

• 13.9 impressions per occupied cell (2.0e10 / 1.44e9).
• 0.139 clicks per occupied cell (2.0e8 / 1.44e9), roughly seven cells per click.

Granularity, or grain, is how finely time is sliced: minute, hour, or day.

At minute grain, 13.9 impressions collapse into each cell, so storing impressions as cells compresses them 13.9x. Clicks do not compress at all: at 0.139 clicks per cell, a click-only table would store more rows than it has events. 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.

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

How much serving state that is depends entirely on how long you keep each grain. Three independent stacks, and the coarse tiers turn out nearly free (the cell count shrinks with the grain because “480 minutes” becomes 8 hours or 1 day):

That totals 721.5 GB, or 2.16 TB with RF 3.

545 TB of raw log against 2.16 TB of serving state is 252x. Two techniques produce that ratio:

• A rollup precomputes coarser summaries from finer ones, so an hourly number is stored instead of 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 millionths of a dollar), never as a floating-point number.

The obvious reason would be precision, so dispose of that first. The worst-case relative error of naively summing n floats is about n x eps, where eps (machine epsilon, the smallest relative gap between two doubles) is ≈ 1.1e-16. A full day of clicks gives 2.0e8 x 1.1e-16 ≈ 2.2e-8, four orders of magnitude inside the 1e-3 budget. Precision is a non-issue.

The real reason is determinism: the same inputs must produce the same output bit for bit. 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. The reconciler would then see the two paths disagree in the low bits on every advertiser, every day, for no reason at all, and a reconciliation that fires on summation order is one you turn off within a week. Integer addition is associative, so the two orders produce bit-identical totals and any diff the reconciler reports is a real one.

API sketch

Five methods. Two details carry the design: query takes an explicit grain instead of inferring one, and billing_snapshot is a separate method, not a flag on query, because the billed number carries different guarantees from a dashboard read.

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

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

The pipeline these methods sit on top of:

flowchart TB
    C(["edge collectors<br/>global, 233k events/s avg<br/>701k at peak x3"]) --> L["append-only log (system of record)<br/>partitioned, RF 3"]
    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 (rebuildable)<br/>upsert on cell key<br/>2.16 TB")]
    A --> AM["late / IVT amendments<br/>restatement stream"]
    AM --> S
    L --> AR[("raw archive (authoritative)<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 (authoritative, immutable)<br/>only number anyone may invoice"]
    R -->|"outside threshold"| H["hold invoice<br/>page the on-call"]
    S --> Q["query tier<br/>+ 1 s TTL cache"]

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. The three authoritative copies (log, archive, billing snapshot) are deliberate: the log and its long-lived archive are the system of record, and the snapshot is the only number anyone may invoice. Everything else is a materialization that can be deleted and rebuilt from the log.

Follow one click through the pipeline

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

Dedup and enrich. This stage keeps an event_id set (a table of the unique identifiers it has already seen) with a 24-hour TTL (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 (online analytical processing, a database built for aggregating over huge numbers of rows instead of updating single ones). Each write is an upsert: it inserts the row if missing and overwrites it if present, keyed by the cell.

Amend. Events that arrive too late for their window take a second path 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 that can be deleted and rebuilt from the log. That property is what makes the recount possible at all, and it is why the archive, not the OLAP store, is the thing you protect.

Deep dive

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

Windowing, and why event time is not a preference

Windowing looks like a matter of preference, but both choices are forced: only one of three window shapes may touch a billable number, and the timestamp that decides a window must be the one written on the device, not 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. It groups one user’s activity until a gap of inactivity ends it, so its width depends on the data.

Sliding windows are the reflex answer, and they cost five times as much: each event lands in 5 windows, so 5 stored cells instead of 1, which turns the 1.44e9-cell grid into 7.2e9 cells and the storage into 360 GB/day. That is five bytes written for every byte the question needs, to answer something you can answer with five reads.

So store tumbling minutes and sum five adjacent cells at query time. That moves the cost from the write path to the read path. Read fan-out (the number of stored rows one query touches) is the cheap resource here: the query tier serves 2,000 queries/s against 6,945 clicks/s at peak, so a 5x read multiplier lands on the smaller of the two rates. (This says nothing about the full 701,389 events/s firehose, which reads never approach.)

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

Event time versus processing time

This is the decision everything else depends on. 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, the second row here:

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, and under processing-time windows those two runs bucket the same event differently. Their disagreement then carries no information, and the whole reconciliation apparatus is theatre. Event time is not a quality bar; it is the precondition for having one.

What event time costs: state

The cost of event time is state: the data the aggregator holds in memory between reading an event and publishing a result. A window cannot be released until you stop expecting events for it, so every window still open is state you carry, and the longer you tolerate lateness, the more windows stay open at once.

• Tolerating 30 s of lateness keeps about 2 minutes of cells in flight: 1e6 cells/min x 2 x 50 B ≈ 100 MB, which fits in ordinary process memory.
• Tolerating 24 h keeps a whole day of cells: 1.44e9 x 50 B ≈ 72 GB, which does not.

72 GB means an on-disk state backend, which drags in three more problems: checkpointing (periodically writing the whole state somewhere durable so a crash does not lose it), the recovery time to reload it, and 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.

The watermark, priced

The aggregator 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 lesson turns on: no single value is good enough.

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 it passes T, every window ending before T may be closed and published.

It is a guess, and the honest way to make a guess is to price both sides:

• Set it too early and you close windows before the stragglers arrive, throwing 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.

The measured input: event-time skew

Everything here rests on one measured input. Event-time skew is the delay between when an event happened and when the pipeline can see it: network transit, retries, and, dominantly, mobile SDKs (software development kits, the libraries an app embeds to report events) that buffer while offline and flush on reconnect.

The distribution of that skew, measured over a week and cumulative:

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 %

By 30 s, 98.5% have arrived; the remaining 1.5% is a slow tail that needs a full 24 h to reach 99.999%. Every dollar figure below is a function of this curve alone.

Pricing both sides of the choice

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

Side B: what waiting costs. Budget pacing is the control loop that stops showing an advertiser’s ads once they hit their daily budget; it reads the aggregate to decide. If the aggregate is w seconds stale, a campaign already at its cap keeps being served for w more seconds, and that overdelivery is not billable.

To put a rate on side B: 30% of spend comes from campaigns that exhaust their daily budget, so capped spend is 0.30 x 90M = $27M/day, or $312.50/s as a day average. Each capping campaign overshoots by its own spend rate times w, so overdelivery is w x 312.50 dollars/day. That day-average rate is a lower bound (a campaign that caps by midday was spending at twice the average), so every total below is a floor.

Both sides together, in dollars per day (dropped + overshoot = total):

The total is U-shaped: dropping dominates at small w (8% of revenue at 5 s), overshoot dominates at large w ($1.1M at an hour). At the day-average rate the minimum sits near 5 minutes, at $273,750/day, which is 273,750 / 90M = 0.30%, against a 0.10% budget.

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

Because $312.50/s is a floor, re-run the curve at 2x and 4x the capping rate:

The conclusion survives every row: no single watermark reaches 0.1% under any rate. The headline figures do not survive (the optimum moves to 2 minutes at 4x, and 0.30% is a floor), so quote the error as at least 3x over budget, never exactly 3x. This is the load-bearing result of the lesson: the optimum still misses the target by at least 3x, so a single-horizon design cannot work and you need restatement.

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.

And every number here is a function of the skew distribution and nothing else. Shorten the tail and a single watermark becomes sufficient, deleting the amendment path entirely; lengthen it and the billing date itself has to move. The assumption ledger records this as the load-bearing assumption for exactly that reason.

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 (the message-queue lesson covers why). A sender that gets no acknowledgement cannot tell a lost message from a lost acknowledgement, so it must either resend, risking a duplicate (at-least-once), or not resend, risking a loss (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.

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 (the ID-generator lesson if you want it sortable and smaller). Everything downstream keeps a set of the identifiers seen within a dedup window (the span over which duplicates are considered possible, 24 hours here) and silently drops repeats.

Sizing that set, an open-addressed hash table (entries stored directly in one flat array) at load factor 0.5 (the array is twice the size of the data), with 24 bytes per entry (16-byte key + 8 bytes slot metadata):

• Clicks only: 200M ids x 24 B / 0.5 = 9.6 GB, fits in memory on one machine.
• Both streams: 20.2e9 x 24 B / 0.5 = 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 seen, but occasionally says yes about something it has not: a false positive, at rate p. At p = 1e-6 the standard optimal sizing -ln(p) / (ln 2)^2 gives 28.75 bits (3.59 B) per element, so all 24 hourly filters for both streams cost 72.5 GB, 13x cheaper than the 970 GB exact set.

The honest part: a Bloom false positive discards a real event as a duplicate, a silent undercount. Even applied to the money stream that would be 200M x 1e-6 = 200 clicks/day = $90/day, a thousand times inside budget. You still do not use it on clicks, not because of the rate, but because the auditor’s question is “which 200 clicks?”, and a bit array cannot enumerate what it dropped. So: exact dedup on clicks (where you must be able to name the missing rows), probabilistic dedup on impressions.

The second idempotence, which is free

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

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 must compute a window deterministically from a slice of the log it can re-read, and given that, replaying the slice writes the same n again. A duplicated output is then harmless, and the dedup store only has to defend against duplicated input. Making the write a SET instead of 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 results are written. Here the processor could commit two things as one indivisible unit (how far it has read from the log, and the writes it produced), 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 (a widely used stream engine; 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).

The mechanism is real, and it costs three things:

Pick A. The idempotent absolute-value upsert plus a bounded dedup set is cheaper, survives a sink with no transaction support, and makes replay (deliberately re-reading and reprocessing a slice of the log) a supported everyday operation instead of 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.

Design A in miniature: dedup_and_aggregate folds one batch into cells, skipping ids already seen; emit writes with upsert, never increment:

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.

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

    A replay of one log slice must reset BOTH `seen` and `cells`: keeping
    `seen` drops the window silently, keeping `cells` doubles it.
    """
    if seen and not cells:
        raise ValueError("`seen` survived into empty `cells`: 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, not a detail. seen is scoped to one window and discarded with it, and a replay must reset both seen and cells. Get it half-right and you fail silently 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.

Neither is loud on its own, which is why the guard in dedup_and_aggregate raises instead of returning an empty window, and why replay resets are worth testing in both directions.

Lambda, kappa, and the honest pick

The watermark priced ended with an unmet budget: the best single watermark misses 0.1% 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. The fatal objection is not 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 instead of by anything about the data. You have built a measuring instrument whose noise floor is its own construction.

Kappa, and why it is not quite enough

The kappa architecture is the reply: one code path, streaming only, and 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 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 horizons: the same stored cells, consulted at two moments, with two guarantees attached:

Events arriving after w are neither dropped nor held in aggregator state (holding them was the 72 GB). They go 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 with a restatement record saying what changed and why.

flowchart LR
    A["windowed aggregator"] --> C[("stored cells")]
    LATE["late events, past 30 s"] --> AM["amendment stream"]
    AM --> C
    C --> F["fast read at w = 30 s<br/>dashboards, budget pacing"]
    C --> FIN["final read at T+24 h<br/>invoices, advertiser API"]

Two costs survive this design: pacing is still 30 s stale (30 x 312.50 = $9,375/day of overshoot) and a sliver of events is still missing at the cut (0.00001 x 90M = $900/day). Together that is $10,275/day = 0.011%, against a 0.1% budget, 8.8x of margin, where the best single watermark was at least 3x over. The payoff comes from closing early, not late: waiting less costs almost nothing because the amendment path recovers what the early close missed.

The amendment path’s volume is the tail beyond 30 s: 1.5% of events, or 0.015 x 20.2e9 = 303M/day = 3,507/s. That is 1.5% of the ingest rate, one event for every 67 the main path handles, which is why the amendment path can afford to be simple, sequential, and heavily instrumented.

The pick: kappa with restatement, plus a batch recount that is an audit, not 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

All three costs come from one source: a published number is now allowed to change after publication.

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) 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, with each server appearing at many points (virtual nodes) so adding or removing a server moves only a small share of keys. But the ring balances the keyspace: a single key has one hash, lands in one arc, and belongs to one owner, no matter how many virtual nodes you configure. The consistent-hashing lesson derives this.

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, 0.10 + 0.90/64 = 0.1140625, against a uniform share of 1/64 = 0.015625. That is 7.30x on one partition, and no number of virtual nodes moves it, because the hot thing is the aggregation key itself, not where it is placed. You must provision every machine for whatever the busiest one does, so 7.30x skew means paying for 7.3x the hardware to do 1x the work.

The fix: salt the hot key

Salt the hot key: attach a small extra component so one logical key becomes S physical ones, spread across S partitions. With S = 16, the hot advertiser’s 10% splits into sixteen pieces of 0.625% each, so the hottest partition now carries 0.00625 + 0.90/64 = 0.0203125, a skew of 1.30x. 7.30x -> 1.30x, for a fleet that is now nearly balanced.

The salt is derived from event_id through a stable hash, not from a counter or random number, and that 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.

“Deterministic” has to mean across processes, which 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 then re-partitions, and the batch recount disagrees with the stream by construction: a failure the reconciler cannot 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 must add them back together. That sounds expensive and is not. Only keys above the uniform share are worth salting, which caps how many advertisers qualify at 1 / 0.015625 = 64. Each owns roughly 1M / 100k = 10 ads, live in 3 countries, so 30 cells each; split S = 16 ways across 64 advertisers that is 64 x 30 x 16 = 30,720 partial rows per minute, or 512 rows/s into the merge. A laptop could run it.

The unit matters: a cell is (ad, minute, country), not (advertiser, minute). A hot advertiser owns about ten ads in three countries, so the merge sees 30 rows per salt per minute, not one; getting that wrong understates the merge by 30x. And salting only the provably-hot keys is the point: salting everything would multiply the entire 1.44e9-cell grid by 16.

Two pieces of machinery make “provably hot” work:

• A rolling heavy-hitters sketch on the ingest path detects which keys are hot: a small fixed-size structure tracking approximately which keys are most frequent, without 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: one of the S partial aggregates never gets merged, producing a single-account shortfall of roughly 1/S, which shows up in reconciliation as one advertiser short by ~1/16.

The salt and the merge (_stable_hash is deliberately not Python’s hash()):

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

Two other kinds of skew get conflated with this one:

• A hot cell (one ad, one minute, a sudden spike) is the same problem at a finer key, and the same salt handles it.
• A hot query (everyone loading the same dashboard at once) is not a write problem at all. The 1-second cache in the architecture diagram serves every repeat request from one stored answer until it expires.

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 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 uses two thresholds at different levels: a per-advertiser daily alert at 0.05%, and an aggregate daily alert at 0.01%. The aggregate is five times tighter, which looks backwards (the aggregate covers more data) but is deliberate.

The reason is how independent random errors combine: they partly cancel, so summing n of them grows the total absolute error by sqrt(n), not n, and dividing by a total that grew by n shrinks the relative error by sqrt(n). If per-advertiser discrepancies were independent noise across 100,000 advertisers, the aggregate you would expect is 0.0005 / sqrt(100,000) = 0.00000158, 63x 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 pushing every advertiser the same direction: a dropped partition, a timezone boundary, a schema change. The two thresholds look for two different failures (the per-advertiser one for localized bugs, the aggregate one for systematic ones), and 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 added to next month’s invoice; an overbill is a refund, a credit memo, and a trust problem. The asymmetry from 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 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 dedup-window bug.

What neither number can catch

Both computations read the same log, so any bug upstream of the log makes both 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: the edge collector emits a per-minute count of what it sent, and the pipeline compares that 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 an already-published cell. A pipeline with no amendment path has only two bad options: bill the advertiser for fraud, or delay every invoice until the IVT window closes. That is why the amendment path is a functional requirement, not an optimization.

Bottlenecks and scaling

What binds first at each scale is a different question from what is expensive. Three terms first:

• Shuffle: the network step that moves every event with the same key onto the same machine so it can be aggregated there, 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 in progress, exactly one does the work and every waiter gets that answer.

The multi-region case 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 salt-then-merge for free: a region is just a salt whose value you did not get to choose.

Failure modes

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. 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 it: the timestamp looks valid.

Alternatives rejected

A rejection with a number attached is worth more than a preference. Three terms appear in the table:

• Relative standard deviation (sd): the typical size of an estimate’s error as a fraction of the quantity. 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.
• OLTP: online transaction processing, a database tuned for many small reads and writes of individual rows, the opposite of the OLAP workload this pipeline serves.

The assumption ledger

Every design is a set of assumptions with a diagram attached, and the diagram is only correct relative to them. 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.
• Load-bearing: if it is wrong the design is not suboptimal, it is invalid. A box appears or disappears, instead of the count inside a box changing.

The one-line test: 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 last column says what you would build instead if the assumption turned out false.

Conclusion

The load-bearing ideas, in order:

• The output is an invoice line, so “eventually roughly right” is not a design point. A 0.1% miscount is $90,000/day, $32.9M/yr, and $9,000/day on the largest single account.
• Window by event time, not processing time. Only event time makes replay bit-identical, and without that a batch recount and a stream cannot be compared, so reconciliation is meaningless.
• No single watermark meets the budget. Priced on both sides, the optimum still misses 0.1% by at least 3x. Close fast (30 s) and repair with an amendment path instead: total error 0.011%, 8.8x inside budget.
• Exactly-once is at-least-once plus an idempotent write. A 9.6 GB dedup set on clicks, plus emitting absolute cell values (upsert, never increment) so replaying a log slice is a no-op.
• A hot advertiser is an aggregation-key problem, not a placement problem. Consistent hashing leaves 7.30x skew; a deterministic salt on event_id brings it to 1.30x, with a cheap second merge stage.
• The billed number is the recount, not the stream. A batch job over the immutable log is an audit nobody queries; the reconciler holds any invoice it cannot match, and the log is the one artifact you protect.

One line to remember: you never bill the stream, you bill the recount from the immutable log; every other copy is a materialization you can delete and rebuild.

Further reading

• Akidau et al., “The Dataflow Model” (VLDB 2015): event time, watermarks, and windowing, the foundation for the watermark pricing here.
• Tyler Akidau, “Streaming 101” and “Streaming 102”: accessible companions to the same model.
• Jay Kreps, “Questioning the Lambda Architecture”: the case for kappa over lambda that this lesson’s read-horizon design builds on.
• Burton Bloom, “Space/Time Trade-offs in Hash Coding with Allowable Errors” (1970): the original Bloom filter.
• Flajolet et al., “HyperLogLog” (2007): the distinct-count sketch, and why it is wrong for a billable count.