#!/usr/bin/env python3 """Hands-on P07 — a stream processor, assembled from seven lego blocks.""" import random, statistics from collections import defaultdict from _harness import block, run_all rng = random.Random(7) def make_stream(n=20000, span=600_000, tail=0.08): """(event_time, proc_time, key). 8% of events arrive very late.""" ev = [] for _ in range(n): et = rng.uniform(0, span) d = rng.expovariate(1/2000) if rng.random() > tail else rng.uniform(0, 90_000) ev.append((et, et + d, rng.choice("abcd"))) ev.sort(key=lambda e: e[1]) # the system sees PROCESSING-time order return ev @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} @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} @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} @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} @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} @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} @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 {} 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.") if __name__ == "__main__": run_all(assembly, "HANDS-ON P07 — Stream processing, block by block")