Stream Processing

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.

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

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.