--- title: 'Stream Processing' source: 'https://academia.sh/en/courses/data-pipelines/stream-processing' course: 'Data Pipelines and Orchestration' language: en updated: '2026-10-21T05:02:58+00:00' license: 'CC BY-SA 4.0' objectives: - 'Changes the closing tolerance of the period window and counts the dropped rows and polluted summary cells on a late-arriving record.' - 'Writes the freshness cost of the window tolerance by reading the period in which the summary cell was written.' - 'Measures the state carried in stream mode as entries and compares how it grows with the tolerance.' - 'Shows that state lost without a checkpoint turns into rows that cannot be computed and missing summary cells.' --- # Stream Processing Turning the window decision and the carried state into numbers: when the period window is closed the moment the period ends, 40 records that arrive one period late are dropped and 10 of the 55 summary cells are polluted; when the window is kept open one period longer the dropped rows fall to 0, but the 2026-07 cell is written in 2026-08 and the carried state grows from 400 entries to 600. Tolerance is a bet on delay: when the same records are two periods late, tolerance 1 gives exactly the same result as tolerance 0. In none of the nine setups do the five served values change, only the number of polluted cells does. If the window never closes the state grows to 2400 and the mode turns back into batch processing; with the state lost without a checkpoint, 200 rows cannot be computed, the summary falls to 50 cells, and the serving shifts in every region. In the previous three lessons the data was ingested into the pipeline: the method of ingestion, how to ingest under the constraint of the source, and which partition the messages would come from were written down. Whether the ingested row arrives **completely** in the pipeline was not asked. This lesson measures the first form of loss, and that form is not a transport error: it is **a record that arrives late to a window that has closed**. Batch processing has no window question, because the period is looked at after it has closed. In stream processing the source never ends; to take a total, the stream has to be cut somewhere. The cutting decision determines three things at once: which record enters the computation, in which period the result is written, and how much **state** the run carries. This lesson puts those three numbers side by side. - **DI41.** The code is **fictional**; no real stream processor, messaging system, or orchestrator is called. The source is the fictional meter readings of the Data Engineering and MLOps curriculum: 200 subscribers, 12 periods, 2400 rows, seed 20260218. - **DI42. The comparison of batch and stream mode is not repeated**; it was measured in the Batch and Stream Processing lesson of the Introduction to Data Engineering course. The baseline here is the stream mode of that lesson; the only thing added is **how long the window stays open**. - **DI43.** The window is one period. The **watermark** points to the latest period counted as closed; **tolerance** is how many more periods the window stays open before it closes. - **DI44.** A late-arriving record is counted not as "never arrived" but as **arrived one period later**: the forty readings of period 2026-07 arrive in 2026-08. The corruption class is the `late_arrival` class of the common definition. - **DI45.** The carried state has two parts: the rows waiting in open windows and the last index per subscriber. The measured size is the **largest value** the sum of the two reaches during the run. - **DI46.** In stream mode consumption is computed from the state: the consumption of a row is subtracted from the last index of the same subscriber that sits in the state. A row with no counterpart in the state cannot be computed. - **DI47.** Delay is written as **periods**, not as a raw duration. The step of the summary cell ratio is 1/55, that is 0.0182; no ratio is printed on the serving, the value itself is written. - **DI48.** A checkpoint is the storage of the carried state outside the run. Its procedure is not written in this lesson; only the **cost of losing it** is counted. ## Window, Tolerance, and State The block below sets up the common definition and adds a single runner to it. `stream_window` takes in each period the rows that **arrive** in that period, drops those that belong to a window the watermark has closed, keeps the rest waiting in the open windows, and closes the window whose turn has come and writes its summary cells. ```python # FICTIONAL. No real stream processor , messaging system or orchestrator is called; # the meter readings of the measurement network are fictional , seed 20260218. SEED, M32 = 20260218, 0xFFFFFFFF def rng(t): x = ((t ^ (t >> 16)) * 2246822507) & M32 x = ((x ^ (x >> 13)) * 3266489909) & M32 s = [(x ^ (x >> 16)) & M32] def next_value(): s[0] = (s[0] * 1664525 + 1013904223) & M32 return s[0] / 4294967296 return next_value REGION = ["east", "west", "north", "south", "central"] TYPE = ["residential", "commercial"] PERIODS = [f"2026-{a:02d}" for a in range(1, 13)] SUBSCRIBERS = {} for i in range(200): r = rng(SEED + 37 * i) SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)], "start": 10000 + int(r() * 60000)} def generate_source(): # one reading per subscriber per period records = [] for i, (sub, a) in enumerate(SUBSCRIBERS.items()): e = a["start"] for p in PERIODS: r = rng(SEED + 101 * i + 7 * PERIODS.index(p)) e += 4 + int(r() * 46) records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"], "index": e, "unit": "m3"}) return records def s1_screen(records): REQUIRED = ("subscriber", "period", "region", "index", "unit") keep = [k for k in records if all(a in k for a in REQUIRED)] return keep, len(records) - len(keep) def s2_dedupe(records): seen, keep = set(), [] for k in records: key = (k["subscriber"], k["period"]) if key in seen: continue seen.add(key) keep.append(k) return keep, len(records) - len(keep) def s3_consumption(records): # batch mode: consumption from the period difference table = {(k["subscriber"], k["period"]): k for k in records} out, dropped = [], 0 for k in records: i = PERIODS.index(k["period"]) if i == 0: continue prev = table.get((k["subscriber"], PERIODS[i - 1])) if prev is None: dropped += 1 continue out.append(dict(k, m3=int(k["index"]) - int(prev["index"]))) return out, dropped def s4_summary(records): totals = {} for k in records: key = (k["region"], k["period"]) totals[key] = totals.get(key, 0) + k["m3"] return totals, 0 def s5_serve(summary): # per-region mean over the periods region = {} for (b, d), v in summary.items(): region.setdefault(b, []).append(v) return {b: round(sum(v) / len(v), 2) for b, v in region.items()}, 0 def with_arrival(records, delay): """Gives each row an arrival period. late_arrival class: the A001..A040 readings of period 2026-07 arrive `delay` periods later.""" data = [] for k in records: i = PERIODS.index(k["period"]) g = delay if k["period"] == "2026-07" and k["subscriber"] <= "A040" else 0 data.append(dict(k, arrival=PERIODS[i + g])) return data def stream_window(data, tolerance, reset_at=None): """The period window stays open for `tolerance` more periods , then closes. reset_at: the carried state is lost at the start of this period (no checkpoint).""" buffer, last_index, cells, written = {}, {}, {}, {} dropped, state, uncomputable = 0, 0, 0 def close(p, i): nonlocal uncomputable v, _ = s1_screen(buffer.pop(p, [])) v, _ = s2_dedupe(v) c = [] for k in v: if k["subscriber"] in last_index: c.append(dict(k, m3=k["index"] - last_index[k["subscriber"]])) elif PERIODS.index(p) > 0: uncomputable += 1 for k in v: last_index[k["subscriber"]] = k["index"] # state: last index per subscriber o, _ = s4_summary(c) if o: cells.update(o) written[p] = PERIODS[i] for i, d in enumerate(PERIODS): if d == reset_at: last_index.clear() for k in data: if k["arrival"] != d: continue if PERIODS.index(k["period"]) < i - tolerance: dropped += 1 # the watermark closed the window continue buffer.setdefault(k["period"], []).append(k) state = max(state, sum(len(v) for v in buffer.values()) + len(last_index)) if i - tolerance >= 0: close(PERIODS[i - tolerance], i) for j in range(max(0, len(PERIODS) - tolerance), len(PERIODS)): close(PERIODS[j], len(PERIODS) - 1) return {"dropped": dropped, "state": state, "cells": cells, "written": written, "uncomputable": uncomputable} raw = generate_source() v, _ = s1_screen(raw) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) BASELINE, _ = s4_summary(v) def dirty(h): return sum(1 for name in set(h) | set(BASELINE) if h.get(name) != BASELINE.get(name)) def diff(h): s, t = s5_serve(h)[0], s5_serve(BASELINE)[0] d = {b: round(s[b] - t[b], 2) for b in t if round(s.get(b, 0) - t[b], 2)} return d if d else "empty" print("baseline: consumption rows", len(v), ", summary cells", len(BASELINE)) print("baseline serving", s5_serve(BASELINE)[0]) print() print("40-row delay tolerance dropped state dirty/55 07 written serving diff") for delay in (0, 1, 2): for t in (0, 1, 2): r = stream_window(with_arrival(raw, delay), t) print(f"{delay:>10} period {t:>9} {r['dropped']:>7} {r['state']:>6}" f" {dirty(r['cells']):>9} {r['written']['2026-07']:>10} {diff(r['cells'])}") ``` ``` baseline: consumption rows 2200 , summary cells 55 baseline serving {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} 40-row delay tolerance dropped state dirty/55 07 written serving diff 0 period 0 0 400 0 2026-07 empty 0 period 1 0 600 0 2026-08 empty 0 period 2 0 800 0 2026-09 empty 1 period 0 40 400 10 2026-07 empty 1 period 1 0 600 0 2026-08 empty 1 period 2 0 800 0 2026-09 empty 2 period 0 40 400 10 2026-07 empty 2 period 1 40 600 10 2026-08 empty 2 period 2 0 800 0 2026-09 empty ``` ## The Result the Window Decision Changes The first three rows show the **cost** of the decision without any loss. Even when no record arrives late, tolerance 1 raises the carried state from 400 entries to 600, tolerance 2 raises it to 800, and the 2026-07 cell is written one or two periods late. This cost is paid **in every run**; whether a late record arrives or not does not change it. Keeping a window open is reserving room for a record that has not arrived, and the room is reserved even if that record never comes. The middle three rows give the return. When the forty readings arrive one period late, tolerance 0 **drops** them: the window has closed and the watermark is not taken back. The result is that **10** of the 55 summary cells are polluted — the 2026-07 cells are left short, and the 2026-08 cells are inflated, because the consumption of those forty subscribers is computed from a two-period difference. A record dropped from one window breaks not only its own cell but **also the cell of the period that follows**; the reason is that consumption is computed from the state. At tolerance 1 the same records still find the window open: dropped rows **0**, polluted cells **0**. The last three rows define what tolerance is: a guess. When the same forty records are two periods late, tolerance 1 drops them too, and the result is **exactly the same** as tolerance 0 — 40 dropped rows, 10 polluted cells. Tolerance is a bet on how long the delay will be; when the bet does not hold, the price paid cannot be taken back, and moreover it was paid before the right answer could be obtained. Tolerance 2 wins this time, but in all of the tolerance 2 rows the state has risen to 800 entries. The last column is the warning of the lesson. In all nine rows the **serving difference is empty**. Despite ten polluted cells, the five region values served are exactly the same as the baseline, because what falls short in 2026-07 is in excess in 2026-08 and the mean of the eleven periods does not move. The only evidence that the window decision is wrong is **the comparison at cell level**; no observation that looks at the output can question this decision. The rows that give the right answer also have a cost, and that cost sits in the same table: with tolerance the 2026-07 cell is written one or two periods late. Keeping the window open is delaying the result. ## The Two Costs of State Enlarging the tolerance buys correctness up to a point, and that has a limit. The first table of the block below shows the limit: what happens if the window never closes. The second table measures the real nature of the state — the carried state is not a memory size but the **condition for the pipeline to be able to produce the right answer**. In stream mode consumption is computed from the last index in the state; if the state is lost, the row cannot be computed. The second table runs the tolerance 1 setup twice: with the state kept between runs and with the state lost at the beginning of 2026-08. ```python print("window tolerance state dropped dirty/55 07 written") for name, t in (("closes each period", 1), ("never closes", 11)): r = stream_window(with_arrival(raw, 1), t) print(f"{name:<18} {t:>9} {r['state']:>6} {r['dropped']:>8} {dirty(r['cells']):>9}" f" {r['written']['2026-07']:>10}") print() print("state uncomputable cells dirty/55") for name, rs in (("kept", None), ("lost", "2026-08")): r = stream_window(with_arrival(raw, 1), 1, reset_at=rs) print(f"{name:<14} {r['uncomputable']:>13} {len(r['cells']):>6} {dirty(r['cells']):>9}") print(" serving", s5_serve(r["cells"])[0]) ``` ``` window tolerance state dropped dirty/55 07 written closes each period 1 600 0 0 2026-08 never closes 11 2400 0 0 2026-12 state uncomputable cells dirty/55 kept 0 55 0 serving {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} lost 200 50 5 serving {'central': 1325.6, 'north': 1208.7, 'east': 873.7, 'west': 835.2, 'south': 1068.2} ``` The first table gives the upper limit of the tolerance. If the window never closes, no record is dropped and no cell is polluted, but the carried state rises to **2400 entries**, that is the whole of the source, and the 2026-07 cell is written in the last period, 2026-12. This row is no longer stream processing; it is a batch run that accumulates all the data and processes it once at the end. **The window is what makes stream processing stream**: a window that never closes gives back all of the mode's freshness gain. The second table measures the fragility of the state. When the state is lost, **none of the 200 rows** in the 2026-08 window can be computed, because no subscriber's previous index is at hand. The summary falls from 55 cells to **50**: one period is not written at all. The effect of this on the serving is visible this time — all five region values shift, and in two directions at once, with central rising while north falls. The reason is that the region mean is now taken over **ten** periods, not eleven; how the dropped period differs from the mean determines the sign. When the two costs are put side by side, a strange balance of the stream mode appears. The late record corrupted ten cells without moving the serving at all; the lost state moved the serving, but by leaving five cells **missing**, that is by dropping a countable trace. A missing cell is a question that can be asked in an audit; a wrong cell is not. Storing the state outside the run — a **checkpoint** — removes this second cost; it does not remove the first, that is the window decision. What the two costs have in common is that both are **a consequence of the window**. There is no such question in a batch run: every run reads the whole of the source again, a late row enters the computation by itself in the next run, and no state is left to carry. Because the stream mode drops the row it has read, it loses this free correction; in its place it leaves two decisions — how long the window will stay open and where the state will be stored. Neither of these two decisions can be made by looking at the data; they are made by looking at **the distribution of the delay** and the frequency with which the run is interrupted. A pipeline's number is therefore not read from its code alone. ## Summary - In stream processing the window decision determines three things at once: which record enters the computation, in which period the result is written, and how much state the run carries. - At tolerance 0, 40 records that arrive one period late are dropped and 10 of the 55 summary cells are polluted; at tolerance 1 the dropped rows and polluted cells fall to 0, and the price is that the 2026-07 cell is written one period late. - Tolerance is a bet on delay: when the same records are two periods late, tolerance 1 gives exactly the same result as tolerance 0, moreover with a state of 600 entries already paid. - The cost of the tolerance does not depend on the run: the carried state is 400, 600, and 800 entries whether or not a record arrives late; if the window never closes the state rises to 2400, that is the whole of the source. - In none of the nine setups do the five served values change; the only witness of the ten polluted cells is the comparison at cell level. - State lost without a checkpoint makes 200 rows impossible to compute, lowers the summary from 55 cells to 50, and moves the serving in every region. ## Next Step In this lesson the source of the loss was the pipeline's own decision: the record that arrived after the window closed was dropped on purpose, how many were dropped was counted, and when the decision was changed the number fell to zero. The record had **reached** the pipeline; it had only reached it late. What the number would be if a record never arrived, or arrived twice, was not asked. The delivery between the ingestion and the pipeline was assumed to be flawless throughout the lesson: every row sent was processed as if it had arrived exactly once. The next lesson removes this assumption and puts the three delivery modes side by side: the setups in which a row can be lost, can arrive twice, and is written with a key. The question to be measured is this: of loss and duplication, which corrupts the served number, which is caught by a check, and why the **repairability** of the two is not the same.