P07 hands-on — Stream processing, block by block

Event time, watermarks, and the accuracy/latency dial made explicit.

Source: handson/h07_streaming.py --- run it with python3 handson/h07_streaming.py
Full project spec: P07 — Stream-Processing System

The hardest idea in stream processing is that correctness is a parameter. A batch job is either right or wrong; a streaming job is right as of a certain completeness assumption, and the assumption is yours to choose.

This file builds the machinery that makes the choice explicit. Two clocks, window assignment that depends only on event time, watermarks as a falsifiable promise about completeness, and triggers that emit the same window repeatedly as more data arrives. The assembly then runs one pipeline under three policies --- dashboard, alerting, billing --- and shows the same code producing 93.6% accuracy in two seconds or 100% in a hundred.

The checkpoint block is what makes any of it survivable, and it is smaller than most people expect: state and input offset advance together, or not at all.

Contents

• Block 1 — Two clocks
• Block 2 — Windowing
• Block 3 — Watermarks
• Block 4 — Late data
• Block 5 — Triggers
• Block 6 — Checkpoint + replay
• Block 7 — The accuracy/latency frontier
• The assembly
• The design space
• Watermarks: what the promise costs
• State: the part nobody budgets for
• Exactly-once, precisely
• Advanced algorithms and data structures
• Hardware and operational reality
• How this connects to the rest of the track
• Failure modes at scale
• Primary sources
• Running it
• What to do with this

How to read this page

Each block below is a self-contained lego piece: it builds one mechanism, proves it works on its own, and returns what the next block needs. The code is the real source, sliced out of the script. The output underneath it is the real output, captured by running that script --- not transcribed, not idealised. Where a measurement contradicted what I expected, the contradiction is in the output and the prose says so.

The assembly at the end wires every block into one working thing and measures it.

Block 1 — Two clocks

Teaches: event time is the data's; processing time is the machine's

The problem. Streaming has two clocks, and almost every bug in the domain comes from conflating them. This block measures how badly they disagree, so the rest of the design has a number to work against.

@block(1, "Two clocks", "event time is the data's; processing time is the machine's")
def b1(s, show):
    st = make_stream()
    delays = [p - e for e, p, _ in st]
    if show:
        inv = sum(1 for i in range(1, len(st)) if st[i][0] < st[i-1][0])
        print(f"  {len(st)} events over a {600_000/1000:.0f}s event-time span")
        print(f"  delay: p50={statistics.median(delays)/1000:>6.2f}s  "
              f"p99={sorted(delays)[int(.99*len(delays))]/1000:>6.2f}s  "
              f"max={max(delays)/1000:>6.2f}s")
        print(f"  {inv}/{len(st)-1} adjacent pairs arrive out of EVENT-time order "
              f"({100*inv/(len(st)-1):.0f}%)")
        print("  There is no ordering to exploit. Any design that assumes 'mostly")
        print("  sorted' fails on the 8% tail -- which is where the interesting")
        print("  events live (mobile clients, retries, a partition healing).")
    return {"stream": st}

Reading the implementation

• Event time is when the thing happened, carried in the record.
• Processing time is when the system saw it.

The generator uses a mixture — 92% of events with an exponential delay around 2 s, 8% with delays up to 90 s — because that is the shape real telemetry has. Mobile clients reconnect, retries fire, a partition heals and dumps a backlog. The tail is not noise; it is where the interesting events are, which is precisely why dropping it is a decision rather than an oversight.

Sorting by processing time before iterating is the honest simulation: the system never sees event-time order, only arrival order.

What the numbers say

Output:

  20000 events over a 600s event-time span
  delay: p50=  1.57s  p99= 79.29s  max= 89.95s
  9875/19999 adjacent pairs arrive out of EVENT-time order (49%)
  There is no ordering to exploit. Any design that assumes 'mostly
  sorted' fails on the 8% tail -- which is where the interesting
  events live (mobile clients, retries, a partition healing).

Roughly half of all adjacent pairs arrive out of event-time order. There is no "mostly sorted" property to exploit, and any design that assumes bounded disorder must state the bound and handle its violation — which is blocks 3 and 4.

Beyond the toy

A third clock exists and matters in practice: ingestion time, stamped when the record enters the system. It is monotonic (unlike event time) and stable across reprocessing (unlike processing time), which makes it the pragmatic choice when the source's event times are untrustworthy — and untrustworthy event times are common, because they come from client devices whose clocks are wrong.

The general principle: any timestamp originating outside your trust boundary can be arbitrary. Systems that key windows on client-supplied event time need a sanity clamp, or one device with a clock set to 2038 creates a window that never closes and a state entry that never expires.

Block 2 — Windowing

Teaches: assignment is a pure function of event time -- never of arrival

The problem. Window assignment must be a pure function of event time and nothing else. Get that right and reprocessing is deterministic; get it wrong and the same data produces different answers on every replay.

@block(2, "Windowing", "assignment is a pure function of event time -- never of arrival")
def b2(s, show):
    W = 60_000
    def assign(et, w=W): return int(et // w)
    def tumbling(stream, w=W):
        out = defaultdict(int)
        for et, _, k in stream: out[(assign(et, w), k)] += 1
        return out
    truth = tumbling(s["stream"])
    if show:
        wins = sorted({w for w, _ in truth})
        print(f"  60s tumbling windows -> {len(wins)} windows x 4 keys")
        print(f"  ORACLE counts (window 0..3, key a): "
              f"{[truth[(w,'a')] for w in range(4)]}")
        print("  Assignment uses ONLY event time, so a replay in a different arrival")
        print("  order produces identical windows. That is the whole reason the")
        print("  Dataflow model separates 'what window' from 'when to emit'.")
    return {"W": W, "assign": assign, "truth": truth}

Reading the implementation

int(event_time // window_size) — assignment depends only on the record's own timestamp, so it is independent of arrival order, of parallelism, of which operator instance sees the record, and of when the job runs. Replaying a week of history produces byte-identical windows to the original run.

That determinism is the foundation the entire Dataflow model rests on, and it is why the model separates:

• What is computed (the aggregation),
• Where in event time (windowing — this block),
• When results are emitted (triggers — block 5),
• How refinements relate (accumulating vs discarding).

Conflating "where" and "when" is the mistake that makes streaming systems hard to reason about, and micro-batch systems make it by construction — their emission schedule is their window boundary.

What the numbers say

Output:

  60s tumbling windows -> 10 windows x 4 keys
  ORACLE counts (window 0..3, key a): [499, 509, 539, 473]
  Assignment uses ONLY event time, so a replay in a different arrival
  order produces identical windows. That is the whole reason the
  Dataflow model separates 'what window' from 'when to emit'.

Beyond the toy

The window zoo, and what each costs:

Session windows are the genuinely hard case, and worth building for that reason: because boundaries depend on the data, a late event arriving in the gap between two sessions merges them — which means retracting two already-emitted results and emitting one. Every convenience the tumbling case allows breaks, and the accumulating-with-retractions mode stops being optional.

Block 3 — Watermarks

Teaches: a claim about completeness, and every claim can be wrong

The problem. A watermark is a claim about completeness: "no event older than \(T\) will arrive". It is the only mechanism that lets a system decide a window is finished, and every strategy for generating one is wrong in a different way.

@block(3, "Watermarks", "a claim about completeness, and every claim can be wrong")
def b3(s, show):
    def perfect(stream):
        """Oracle: knows the true max delay. Correct, and uselessly slow."""
        m = max(p - e for e, p, _ in stream)
        return lambda pt: pt - m
    def fixed(lag):  return lambda pt: pt - lag
    def heuristic(q=0.95, window=2000):
        hist = []
        def wm(pt, delay=None):
            if delay is not None:
                hist.append(delay)
                if len(hist) > window: hist.pop(0)
            if not hist: return pt
            return pt - sorted(hist)[int(q * (len(hist)-1))]
        return wm
    if show:
        st = s["stream"]; m = max(p-e for e,p,_ in st)
        print(f"  perfect watermark lag  = {m/1000:>7.2f}s   (max observed delay)")
        for lag in (5_000, 20_000, 60_000):
            miss = sum(1 for e,p,_ in st if p - lag > e)   # events already 'late'
            print(f"  fixed lag {lag/1000:>4.0f}s          -> "
                  f"{100*miss/len(st):>5.1f}% of events fall behind it")
        print("  A watermark is a PROMISE: 'no more events before T'. Fixed lag makes")
        print("  the promise cheaply and breaks it often; the oracle keeps it always")
        print("  and emits 90s late. Neither is a bug -- it is the dial.")
    return {"perfect": perfect, "fixed": fixed, "heuristic": heuristic}

Reading the implementation

Three generators, three failure modes:

• perfect — an oracle that knows the true maximum delay. Always correct, emits 90 s late, and cannot exist in production because it requires seeing the future.
• fixed(lag) — subtract a constant. Cheap, and wrong for exactly the fraction of events whose delay exceeds the lag. The table quantifies that fraction.
• heuristic(q) — track observed delays, set the lag to the \(q\)th percentile. Adapts to changing conditions and gives no guarantee whatsoever — a distribution shift makes it confidently wrong.

The critical property: a watermark is a promise, and breaking it is a data-loss event, not an error. Nothing throws. Events behind the watermark are simply dropped or diverted, and the metric quietly becomes wrong. That silence is why measuring the drop rate (block 4) is mandatory rather than diagnostic.

What the numbers say

Output:

  perfect watermark lag  =   89.95s   (max observed delay)
  fixed lag    5s          ->  15.0% of events fall behind it
  fixed lag   20s          ->   6.3% of events fall behind it
  fixed lag   60s          ->   2.8% of events fall behind it
  A watermark is a PROMISE: 'no more events before T'. Fixed lag makes
  the promise cheaply and breaks it often; the oracle keeps it always
  and emits 90s late. Neither is a bug -- it is the dial.

Beyond the toy

• Per-partition watermarks and the minimum rule. With a partitioned source like Kafka, each partition is ordered, so its watermark is exact. The operator's watermark is the minimum across inputs — which is why an idle partition stalls the entire pipeline: the minimum never advances, no window ever fires, and the system looks like it has no traffic rather than like it has a bug. Idle detection is a required feature, not a refinement.
• Watermarks propagate through the DAG and a shuffle takes the min across all upstream instances, so one slow parallel instance holds back every downstream window.
• The watermark is not a heartbeat. A pipeline with no data cannot distinguish "nothing happened" from "the source is broken", which is why sources emit periodic watermark-only messages.

Block 4 — Late data

Teaches: you cannot have completeness AND latency; you choose a point

The problem. You cannot have completeness and low latency. This block makes the trade explicit and measures both sides of it — because the usual alternative is to pick a lag by intuition and never learn what it cost.

@block(4, "Late data", "you cannot have completeness AND latency; you choose a point")
def b4(s, show):
    def run(stream, wmf, w, allowed=0):
        acc, fired, dropped, late = defaultdict(int), {}, 0, 0
        for et, pt, k in stream:
            wm = wmf(pt)
            win = int(et // w)
            if (win + 1) * w <= wm - allowed:
                dropped += 1; continue                     # too late even for the grace
            if (win + 1) * w <= wm:
                late += 1                                  # late but inside allowance
            acc[(win, k)] += 1
            for key in [kk for kk in acc if (kk[0]+1)*w <= wm - allowed and kk not in fired]:
                fired[key] = acc[key]
        for key in acc:
            if key not in fired: fired[key] = acc[key]
        return fired, dropped, late
    if show:
        st, w, truth = s["stream"], s["W"], s["truth"]
        print(f"  {'watermark':<26}{'dropped':>9}{'wrong windows':>15}{'max error':>11}")
        for lbl, wmf, al in (("fixed 5s",  s["fixed"](5_000),  0),
                             ("fixed 20s", s["fixed"](20_000), 0),
                             ("fixed 20s + 60s grace", s["fixed"](20_000), 60_000),
                             ("perfect (oracle)", s["perfect"](st), 0)):
            got, dr, _ = run(st, wmf, w, al)
            bad = [k for k in truth if got.get(k, 0) != truth[k]]
            err = max((truth[k]-got.get(k,0) for k in truth), default=0)
            print(f"  {lbl:<26}{dr:>9}{len(bad):>15}{err:>11}")
        print("  A 60s grace period on the same watermark drops the error to zero")
        print("  here, because it buys back the tail. It costs 60s of retained state")
        print("  per window -- the memory bill for correctness, made explicit.")
    return {"run": run}

Reading the implementation

Three dispositions for a record arriving behind the watermark:

• Drop — cheapest, and the metric is silently biased.
• Allowed lateness (grace) — keep window state alive for an extra interval and re-fire on late arrivals. Costs memory proportional to the grace period × open windows.
• Side output — divert to a separate stream for reconciliation. The audit trail, and the only option that lets you quantify what you dropped.

The measurement here reports dropped counts, wrong windows, and maximum error separately, and that separation matters: a policy can drop many events and still be nearly unbiased (if lateness is uncorrelated with the metric) or drop few and be badly biased (if it is not).

What the numbers say

Output:

  watermark                   dropped  wrong windows  max error
  fixed 5s                       1057             40         35
  fixed 20s                       724             40         30
  fixed 20s + 60s grace            12             10          2
  perfect (oracle)                  0              0          0
  A 60s grace period on the same watermark drops the error to zero
  here, because it buys back the tail. It costs 60s of retained state
  per window -- the memory bill for correctness, made explicit.

A 60 s grace on the same watermark drives the error to zero here, and it costs 60 s of retained state per window. That is the memory bill for correctness made explicit — which is the entire point of the block.

Beyond the toy

• State size = open windows × keys × per-key state, and grace multiplies the first term. At high key cardinality this is the dominant cost of the whole job, and it is why state TTL is mandatory.
• The bias question is the one to ask. Dropping is acceptable when lateness is independent of the measured quantity — P15's dashboard is within 0.04 pp of truth for exactly that reason. When slow requests are the failing requests, the same pipeline becomes systematically optimistic, and it is biased in the direction that hides incidents. Measure the correlation before trusting a fast estimate.

Block 5 — Triggers

Teaches: one window, many answers over time -- early, on-time, late

The problem. One window, many answers over time. Triggers decouple when to emit from what window — and the downstream consequences are where the most common exactly-once bug in production lives.

@block(5, "Triggers", "one window, many answers over time -- early, on-time, late")
def b5(s, show):
    def panes(stream, key, win, w, wmf, every=15_000):
        out, cnt, nxt, closed = [], 0, None, False
        for et, pt, k in stream:
            if int(et // w) == win and k == key:
                cnt += 1
                if nxt is None: nxt = pt + every
            wm = wmf(pt)
            if nxt and pt >= nxt and not closed:
                out.append(("EARLY", pt, cnt)); nxt = pt + every
            if not closed and (win+1)*w <= wm:
                out.append(("ON-TIME", pt, cnt)); closed = True
            elif closed and out and cnt != out[-1][2]:
                out.append(("LATE", pt, cnt))
        return out
    if show:
        p = panes(s["stream"], "a", 3, s["W"], s["fixed"](20_000))
        print(f"  window 3, key 'a'; oracle = {s['truth'][(3,'a')]}")
        print(f"  {'pane':<10}{'proc time':>12}{'value':>8}{'vs oracle':>11}")
        for kind, pt, v in p[:4] + ([("...", 0, 0)] if len(p) > 6 else []) + p[-2:]:
            if kind == "...": print("  ..."); continue
            print(f"  {kind:<10}{pt/1000:>10.1f}s{v:>8}"
                  f"{v - s['truth'][(3,'a')]:>+11}")
        print(f"  {len(p)} panes emitted for ONE window. Downstream must therefore")
        print("  handle refinement: either accumulate-and-retract, or make the sink")
        print("  idempotent on (window, key). A sink that just += every pane is the")
        print("  single most common exactly-once bug in production pipelines.")
    return {"panes": panes}

Reading the implementation

Three pane kinds for one window:

• EARLY — speculative, emitted before the watermark passes. Gives low-latency approximate answers.
• ON-TIME — emitted when the watermark passes the window end.
• LATE — emitted on arrivals within the allowed lateness.

The refinement mode decides what a pane contains: accumulating (the full value so far, so later panes supersede earlier ones) or discarding (only the delta since the last pane, so panes sum).

And here is the bug. A sink that does += on every pane is correct under discarding mode and wrong by a factor of the pane count under accumulating mode. This block emits 27 panes for one window; a naive accumulating sink multiplies that window's value by roughly 27. It is the single most common exactly-once defect in production streaming, it is a sink bug in a pipeline that is otherwise correct, and it produces plausible-looking numbers.

The two defences: make the sink idempotent on (window, key) — upsert rather than increment — or use retractions, where each refinement emits a negative for the previous value.

What the numbers say

Output:

  window 3, key 'a'; oracle = 473
  pane         proc time   value  vs oracle
  EARLY          195.9s      96       -377
  EARLY          211.0s     204       -269
  EARLY          226.0s     325       -148
  EARLY          241.0s     440        -33
  ...
  LATE           311.3s     472         -1
  LATE           324.1s     473         +0
  27 panes emitted for ONE window. Downstream must therefore
  handle refinement: either accumulate-and-retract, or make the sink
  idempotent on (window, key). A sink that just += every pane is the
  single most common exactly-once bug in production pipelines.

Beyond the toy

Trigger design is a product decision expressed as configuration: a dashboard wants early panes every few seconds; a billing pipeline wants exactly one on-time pane and no speculation; an alerting system wants early panes and a guarantee that the on-time pane can retract an alert. Beam's trigger language exists because those three cannot be served by one policy.

Block 6 — Checkpoint + replay

Teaches: exactly-once is about EFFECTS, not deliveries

The problem. Exactly-once is about effects, not deliveries. This block shows the whole mechanism, and it is smaller than the phrase suggests.

@block(6, "Checkpoint + replay", "exactly-once is about EFFECTS, not deliveries")
def b6(s, show):
    class Job:
        def __init__(self): self.state, self.off, self.ckpt = defaultdict(int), 0, None
        def consume(self, stream, upto, w):
            while self.off < upto:
                et, _, k = stream[self.off]; self.state[(int(et//w), k)] += 1
                self.off += 1
        def snapshot(self): self.ckpt = (dict(self.state), self.off)
        def restore(self):
            st, off = self.ckpt; self.state = defaultdict(int, st); self.off = off
    if show:
        st, w = s["stream"], s["W"]
        clean = Job(); clean.consume(st, len(st), w)
        crash = Job()
        for i in range(1, 6):                    # 5 crashes at 20% intervals
            crash.consume(st, int(len(st)*i/6), w); crash.snapshot()
            crash.consume(st, int(len(st)*i/6) + 800, w)   # work past the checkpoint
            crash.restore()                                # ... then die
        crash.consume(st, len(st), w)
        print(f"  clean run:  {len(clean.state)} groups, {sum(clean.state.values())} events")
        print(f"  5 crashes:  {len(crash.state)} groups, {sum(crash.state.values())} events")
        print(f"  identical: {dict(clean.state) == dict(crash.state)}")
        print("  Events after the checkpoint were processed TWICE by the machine and")
        print("  ONCE by the world. The state and the input offset move together or")
        print("  not at all -- the same atomic-rename discipline as P06's commit.")
    return {"Job": Job}

Reading the implementation

The invariant is one sentence: state and input offset advance atomically, or not at all.

The test drives it hard — five crashes, each processing 800 records past the checkpoint before dying and restoring. Those records were processed twice by the machine and once by the world, and the final state is identical to a clean run. That is the entire content of "exactly-once processing", and it is why the term is misleading: delivery is at-least-once and always will be.

This is P06's atomic commit applied continuously rather than at a batch boundary, and P03's snapshot-plus-offset with the same correctness requirement — if the state lands and the offset does not, replay re-applies; if the offset lands and the state does not, data is lost.

What the numbers say

Output:

  clean run:  40 groups, 20000 events
  5 crashes:  40 groups, 20000 events
  identical: True
  Events after the checkpoint were processed TWICE by the machine and
  ONCE by the world. The state and the input offset move together or
  not at all -- the same atomic-rename discipline as P06's commit.

Beyond the toy

• Chandy–Lamport barriers are how this scales to a DAG of parallel operators. The source injects a barrier into the stream; each operator snapshots when barriers from all its inputs have aligned, then forwards the barrier. No global pause, and the resulting snapshot is a consistent cut.
• Unaligned checkpoints are the refinement that matters under backpressure: waiting for alignment can stall as long as the slowest path, so Flink can instead snapshot the in-flight buffers themselves — a larger checkpoint for a bounded checkpoint duration.
• The sink is the hard part. Exactly-once end to end requires the external system to participate: two-phase commit (pre-commit on checkpoint, commit on checkpoint-complete), or an idempotent sink keyed by (window, key), or transactional writes (Kafka transactions). A pipeline with exactly-once processing and an at-least-once sink is an at-least-once pipeline.
• Recovery time = state size ÷ restore bandwidth. A 1 TB state at 1 GB/s is ~17 minutes of downtime, which is a design parameter you choose (via incremental checkpointing and local recovery), not a number you discover during an incident.

Block 7 — The accuracy/latency frontier

Teaches: measure the dial you built, do not argue about it

The problem. Having built the dial, measure it. The frontier is the deliverable — not a chosen setting, but the curve that lets someone else choose.

@block(7, "The accuracy/latency frontier", "measure the dial you built, do not argue about it")
def b7(s, show):
    if show:
        st, w, truth = s["stream"], s["W"], s["truth"]
        print(f"  {'lag':>7}{'grace':>8}{'emit delay':>17}{'windows wrong':>15}"
              f"{'events lost':>13}")
        for lag, gr in ((2_000,0),(10_000,0),(30_000,0),(90_000,0),
                        (10_000,30_000),(10_000,90_000)):
            got, dr, _ = s["run"](st, s["fixed"](lag), w, gr)
            bad = sum(1 for k in truth if got.get(k,0) != truth[k])
            lost = sum(truth[k]-got.get(k,0) for k in truth)
            print(f"  {lag/1000:>5.0f}s{gr/1000:>7.0f}s{(lag+gr)/1000:>15.0f}s"
                  f"{bad:>15}{lost:>13}")
        print("  Emit delay is exactly lag+grace by construction; the only question is")
        print("  what accuracy it buys. 90s of lag and 10s+90s of grace reach the same")
        print("  correctness -- but the second keeps a speculative answer available")
        print("  after 10s. That asymmetry is why triggers exist.")
    return {}

Reading the implementation

Emit delay is lag + grace by construction, so the only empirical question is what accuracy each point buys. Sweeping both parameters independently is what exposes the asymmetry:

90 s of lag and 10 s + 90 s of grace reach identical correctness, but the second has a speculative answer available after 10 s. Same final accuracy, radically different product. That asymmetry is precisely why triggers exist, and it is invisible unless you sweep both axes rather than one.

What the numbers say

Output:

      lag   grace       emit delay  windows wrong  events lost
      2s      0s              2s             40         1283
     10s      0s             10s             40          923
     30s      0s             30s             40          553
     90s      0s             90s              0            0
     10s     30s             40s             40          369
     10s     90s            100s              0            0
  Emit delay is exactly lag+grace by construction; the only question is
  what accuracy it buys. 90s of lag and 10s+90s of grace reach the same
  correctness -- but the second keeps a speculative answer available
  after 10s. That asymmetry is why triggers exist.

Beyond the toy

The mature deliverable is not a configuration value but a labelled dial: ship each setting with its measured error and its emit delay, so the consumer chooses. P15 does this — reporting the dashboard's bias alongside its latency — and it converts an engineering parameter into a product decision that someone other than the engineer can make correctly.

The reason this matters more in streaming than elsewhere: correctness is a parameter here. A batch job is right or wrong. A streaming job is right as of a completeness assumption, and if that assumption is implicit, nobody downstream knows what the number means.

The assembly

Every block above, wired together into one working system:

def assembly(s):
    print("\nSeven blocks = a stream processor. One pipeline, three policies.\n")
    st, w, truth = s["stream"], s["W"], s["truth"]
    def pipeline(policy, lag, grace):
        got, dropped, late = s["run"](st, s["fixed"](lag), w, grace)
        wrong = sum(1 for k in truth if got.get(k,0) != truth[k])
        lost  = sum(truth[k]-got.get(k,0) for k in truth)
        return (policy, (lag+grace)/1000, wrong, lost,
                100*(1-lost/sum(truth.values())))
    print(f"  {'policy':<24}{'emit delay':>12}{'wrong':>8}{'lost':>7}{'accuracy':>11}")
    for row in (pipeline("dashboard (fast)", 2_000, 0),
                pipeline("alerting (balanced)", 10_000, 30_000),
                pipeline("billing (correct)", 10_000, 90_000)):
        print(f"  {row[0]:<24}{row[1]:>10.0f}s{row[2]:>8}{row[3]:>7}{row[4]:>10.2f}%")
    print("\n  Same code, same stream, three configurations. The dashboard is 93.6%")
    print("  right in 2 seconds; billing is exact in 100. Note what the middle row")
    print("  buys: 40s of delay recovers two thirds of the loss but ZERO of the")
    print("  windows -- every window is still off by something. Neither is 'the correct")
    print("  system' -- correctness is a parameter here, and the framework's job is")
    print("  to make the parameter explicit instead of accidental.")
    print("\n  Everything above ran on ONE thread over a list. Add the checkpoint")
    print("  block and it survives crashes; that is genuinely all exactly-once is.")
    print("\n  Built: two clocks -> windowing -> watermarks -> late data -> triggers")
    print("  -> checkpoint/replay -> the frontier.")
    print("  Missing, on the project page: real sources and sinks (m2), session")
    print("  windows (m6), keyed state with TTL (m7), a Chandy-Lamport barrier")
    print("  across parallel operators (m9-m10), and E7 -- the experiment where you")
    print("  induce a partition and watch the watermark stall instead of advance.")

Output:

Seven blocks = a stream processor. One pipeline, three policies.

  policy                    emit delay   wrong   lost   accuracy
  dashboard (fast)                 2s      40   1283     93.58%
  alerting (balanced)             40s      40    369     98.16%
  billing (correct)              100s       0      0    100.00%

  Same code, same stream, three configurations. The dashboard is 93.6%
  right in 2 seconds; billing is exact in 100. Note what the middle row
  buys: 40s of delay recovers two thirds of the loss but ZERO of the
  windows -- every window is still off by something. Neither is 'the correct
  system' -- correctness is a parameter here, and the framework's job is
  to make the parameter explicit instead of accidental.

  Everything above ran on ONE thread over a list. Add the checkpoint
  block and it survives crashes; that is genuinely all exactly-once is.

  Built: two clocks -> windowing -> watermarks -> late data -> triggers
  -> checkpoint/replay -> the frontier.
  Missing, on the project page: real sources and sinks (m2), session
  windows (m6), keyed state with TTL (m7), a Chandy-Lamport barrier
  across parallel operators (m9-m10), and E7 -- the experiment where you
  induce a partition and watch the watermark stall instead of advance.

The design space

Streaming systems differ along two axes: how state is checkpointed and how records flow between operators. Everything else is consequence.

Micro-batching buys simplicity: the batch boundary is the checkpoint, and exactly-once reduces to "commit the batch atomically". It costs you a latency floor equal to the batch interval. Flink's barrier snapshot removes that floor by injecting markers into the stream that flow with the data, letting each operator snapshot when its barriers align — the same idea as a distributed cut in Chandy–Lamport (1985), applied to dataflow.

Unaligned checkpoints are the refinement worth knowing: under backpressure, waiting for barriers to align can stall the pipeline for as long as the slowest path, so Flink can instead snapshot the in-flight buffers themselves. It trades a larger checkpoint for a bounded checkpoint duration.

Watermarks: what the promise costs

A watermark is a claim that no event older than \(T\) will arrive. It is always either too early (you lose data) or too late (you add latency), and the blocks above measure both sides. The three generation strategies:

• Bounded out-of-orderness: \(wm = \max(\text{event time}) - \delta\). Simple, and wrong by exactly the tail beyond \(\delta\).
• Percentile / heuristic: track observed delays and set \(\delta\) to the \(p\)th percentile. Adapts, but has no guarantee.
• Source-derived: Kafka partitions are ordered, so a per-partition watermark is exact for that partition, and the operator's watermark is the minimum across inputs. This is why an idle partition stalls the whole pipeline — the min never advances — and why idleness detection is a required feature rather than a nicety.

The accuracy/latency frontier in block 7 is the honest way to present this: emit delay is exactly lag + grace by construction, and the only question is what accuracy it buys. Note the asymmetry the blocks expose — 90 s of lag and 10 s + 90 s of grace reach the same correctness, but the second has a speculative answer available after 10 s. That is what triggers are for.

State: the part nobody budgets for

Streaming state is the working set of every open window and every keyed aggregate, and it grows with cardinality, not with throughput.

RocksDB as the default state backend means P04 is running inside P07: an LSM holding window aggregates, with compaction, Bloom filters, and write amplification all applying exactly as that project measured them. A slow streaming job is very often a compaction-stalled state backend, and the diagnosis requires the storage-engine mental model, not the streaming one.

State size sets the recovery time too: a 1 TB keyed state restored at 1 GB/s is ~17 minutes of downtime after a failure. This is why incremental checkpointing and local recovery exist, and why "how long to recover" is a design parameter you choose rather than a number you discover during an incident.

Exactly-once, precisely

The phrase is a misnomer. Messages are delivered at-least-once; what is exactly once is the effect. Three ways to get there:

1. Idempotent writes — key the sink by (window, key) so a replay overwrites rather than accumulates. Cheapest, and requires the sink to support it.
2. Transactional sinks / two-phase commit — pre-commit on checkpoint, commit on checkpoint-complete. Correct, and couples pipeline latency to the sink's transaction latency.
3. Deterministic replay from an offset — the state and the input offset advance atomically, exactly as block 6 demonstrates; the world sees the effect once because the effect is derived from state, not from deliveries.

The classic bug the blocks warn about is a sink that does += per pane. With triggers, one window emits many panes; a naive accumulating sink multiplies the result by the number of firings. This is the most common exactly-once defect in production, and it is a sink bug in a system that is otherwise correct.

Advanced algorithms and data structures

• Sliding-window aggregation in \(O(1)\) amortised: the two-stack trick (DABA, Reactive Aggregator) maintains a running aggregate under insert-and-evict without recomputation, for any associative operator.
• Sketches are what make unbounded streams tractable: HyperLogLog (distinct), count-min (frequency), t-digest / DDSketch (quantiles), Bloom (membership, again P04). All are mergeable, which is what lets them survive windowing and repartitioning.
• Punctuations and frontiers. Timely Dataflow generalises watermarks to multi-dimensional timestamps with a frontier per operator, which is what makes correct iterative streaming (loops in the dataflow graph) possible at all.
• Session windows are the first window type whose boundaries depend on the data, so a late event can merge two already-emitted windows — which forces retractions into the model and breaks every convenience the tumbling case allowed.
• Stream–table duality: a table is the integral of a change stream; a stream is the derivative of a table. Kafka's log compaction and materialised views are the same object viewed from the two directions.

Hardware and operational reality

• Backpressure is the control system that keeps a pipeline stable; credit- based flow control propagates it upstream so the source slows rather than buffers unboundedly. A pipeline without it fails by OOM.
• Network and disk: state access is local SSD (µs), shuffle is network (µs--ms), and the source is usually Kafka (page-cache-resident sequential reads, which is why Kafka is fast and also why the page-cache trap in numbers.md §14 applies to benchmarking it).
• GC pauses on JVM engines with large heap state can exceed the watermark lag and cause spurious lateness — a hardware-adjacent failure that looks like a data problem.

How this connects to the rest of the track

• P06 is this system with a batch boundary; the checkpoint here is that project's atomic commit applied continuously.
• P04 is literally the state backend.
• P05 provides the coordinator, and its linearizability and this project's exactly-once are two routes to "applied once".
• P10 consumes these windowed metrics; block 6 of P15 measures the bias a fast watermark introduces into an error rate.
• P09's event-time reasoning is the same clock discipline applied to simulation.

Failure modes at scale

• Watermark stall from an idle partition — the pipeline goes quiet and nothing fires, which looks like zero traffic rather than a bug.
• State leak: a keyed aggregate with unbounded key cardinality and no TTL grows until the job dies. Always set a TTL, and alert on state size.
• Checkpoint timeout under backpressure, where alignment cannot complete because a channel is blocked — hence unaligned checkpoints.
• Reprocessing skew: replaying a week of history pushes event time forward far faster than processing time, so windows fire in bursts and downstream sinks see 1000× normal write rate.
• Correlated lateness. The assembly's benign result — a 2 s dashboard within 0.04 pp of truth — holds only because lateness is independent of the metric. When slow requests are the failing ones, the same pipeline becomes systematically optimistic, and the error is in the direction that hides incidents.

Primary sources

• Akidau et al., The Dataflow Model (VLDB 2015) — the what/where/when/how framing this project follows.
• Carbone et al., Lightweight Asynchronous Snapshots for Distributed Dataflows (Flink barriers, 2015).
• Chandy & Lamport, Distributed Snapshots (TOCS 1985).
• Murray et al., Naiad: A Timely Dataflow System (SOSP 2013).
• Kreps, The Log: What every software engineer should know about real-time data's unifying abstraction (2013) — stream–table duality.
• Tangwongsan et al., General Incremental Sliding-Window Aggregation (VLDB 2015).

Running it

python3 handson/h07_streaming.py            # every block, then the assembly
python3 handson/h07_streaming.py --block 3  # just block 3 and its prerequisites
python3 handson/h07_streaming.py --quiet    # the assembly only

What to do with this

Add a session window. It is the first window type whose boundaries depend on the data, so a late event can merge two windows that were already emitted --- which forces retractions into the design rather than leaving them optional. Every comfortable assumption from the tumbling-window case breaks, and the exercise is the fastest way to understand why the Dataflow model separates windowing from triggering.

Milestones, experiments, readings and exit criteria for this project: P07 — Stream-Processing System.