Lesson 05 / 10

Batch and Stream Processing

Running the same pipeline in two modes and counting the trade-off between freshness and detection lag: in batch mode the corruption becomes visible one period later, in the 2026-08 run, in stream mode in the same period, in the 2026-07 window. In return, stream mode raises the number of stages from 5 to 8 and the carried state from 0 to 2600 entries; on the other hand, the rows read fall from 13200 to 2400. The period window fixes the flawed partition check: the missing partition that batch mode never catches is caught at stage 0 in stream mode. For the late-arriving row the trade-off reverses: batch mode takes the row back in the next run and leaves all 55 summary cells correct, while stream mode drops the 40 rows at the watermark and permanently dirties 10 cells, yet the five served values stay exactly the same in both modes.

Published October 14, 2026 Updated October 21, 2026

In this lesson

Runs the same pipeline in batch and stream mode and counts, separately for each mode, after how many periods the corruption is seen.
Compares the number of stages stream mode adds and the state entries it carries with the number of rows batch mode reads.
Shows why the period window fixes the flawed partition check.
Compares, for the late-arriving row, the backfill of batch mode with the summary cells that stream mode permanently dirties.

Contents

The previous lesson extracted the profile of the three source classes with all twelve periods at hand: the corrupted period was sitting inside the pipeline and could be compared with the remaining eleven periods. A pipeline does not work like that; it is tied to a schedule, and the schedule determines when the corrupted period will be looked at.

This lesson runs the same pipeline in two modes. Batch processing (batch processing) runs after a period closes and processes every row that has arrived up to that moment from the beginning; stream processing (stream processing) processes a row in the period in which it arrives and carries state to the next period. Both use the same five stages, the same six corruption classes and the same checks; their only difference is the window.

DL47. The code is fictional; no real stream system, queue product or orchestrator is called. The source is the fictional meter readings of the previous lesson: 200 subscribers, 12 periods, 2400 rows, seed 20260218.
DL48. The queue, retry and latency procedures are not repeated; they were measured in the Caching, Queues and Asynchronous Processing course of the Backend Development curriculum and in the Application Layer and Service Interaction course of the System Design and Distributed Systems curriculum. The only thing added here is in which period the corruption is seen.
DL49. In batch mode a run processes every row that has arrived up to that moment from the beginning; in stream mode a row is processed in the period in which it arrives, and the watermark (watermark) drops a row whose window has closed.
DL50. In this lesson a late-arriving row counts not as “never arrived” but as arrived one period later. The corruption class is the same; the only thing added is the row’s arrival period.
DL51. The same five checks are enabled in both modes: schema, type, uniqueness, partition and distribution. The partition check is the flawed one: it looks at the region set and does not see the period.
DL52. Lag is written in two units: periods and stages. The step of the summary cell ratio is 1/55, that is, 0.0182.

Two Modes, One Pipeline

The block below sets up the shared definition and adds two runners. batch processes, at the start of each period, the rows that have arrived up to that moment from the beginning and carries nothing. stream processes each period in its own window and carries two things: the set of keys seen and the last index per subscriber: information the deduplication and consumption stages find for free in batch mode.

# FICTIONAL. No real stream system, queue or orchestrator is called; the counter
# 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


CORRUPTIONS = ["late_arrival", "duplicated", "schema_changed", "unit_changed", "missing_partition", "type_drift"]


def corrupt(records, cls, period="2026-07"):
    """Corrupts a single period , the remaining periods stay intact."""
    out, counter = [], 0
    for k in records:
        if k["period"] != period:
            out.append(k)
            continue
        if cls == "late_arrival":
            if k["subscriber"] <= "A040":           # 40 subscribers never arrived this period
                counter += 1
                continue
            out.append(k)
        elif cls == "duplicated":
            out.append(k)
            if k["subscriber"] <= "A060":           # 60 records arrived twice
                out.append(dict(k))
                counter += 1
        elif cls == "schema_changed":
            y = dict(k)
            y["index_value"] = y.pop("index")       # field name changed
            counter += 1
            out.append(y)
        elif cls == "unit_changed":
            y = dict(k)
            y["index"] = k["index"] * 1000          # liters instead of m3
            y["unit"] = "m3"                        # LABEL DID NOT CHANGE , silent corruption
            counter += 1
            out.append(y)
        elif cls == "missing_partition":
            if k["region"] == "north":              # one region's partition was never written
                counter += 1
                continue
            out.append(k)
        elif cls == "type_drift":
            y = dict(k)
            y["index"] = str(k["index"])            # number turned into text
            counter += 1
            out.append(y)
    return out, counter


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):
    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
        try:
            f = int(k["index"]) - int(prev["index"])
        except (TypeError, ValueError):
            dropped += 1
            continue
        out.append(dict(k, m3=f))
    return out, dropped


def s4_summary(records):
    totals = {}
    for k in records:
        totals[(k["region"], k["period"])] = totals.get((k["region"], k["period"]), 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


STAGES = [("screen", s1_screen), ("dedupe", s2_dedupe), ("consumption", s3_consumption),
          ("summary", s4_summary), ("serve", s5_serve)]


def c_schema(data, fields=("subscriber", "period", "region", "index", "unit")):
    return any(any(a not in k for a in fields) for k in data)


def c_type(data):
    return any(not isinstance(k.get("index"), int) for k in data)


def c_uniqueness(data):
    key = [(k["subscriber"], k["period"]) for k in data]
    return len(key) != len(set(key))


def c_partition_completeness(data):
    """Looks at the region set , does not see the period: flawed over the whole history , not in the window."""
    return set(k["region"] for k in data) != set(REGION)


def c_distribution(summary, baseline, tolerance=0.25):
    for key, v in summary.items():
        t = baseline.get(key)
        if t and abs(v - t) / max(abs(t), 1) > tolerance:
            return True
    return False


CHECKS = [("schema", 0, lambda v, t: c_schema(v)),
          ("type", 0, lambda v, t: c_type(v)),
          ("uniqueness", 0, lambda v, t: c_uniqueness(v)),
          ("partition", 0, lambda v, t: c_partition_completeness(v)),
          ("distribution", 4, lambda v, t: c_distribution(v, t))]


def full_pipeline(records):
    v, _ = s1_screen(records)
    v, _ = s2_dedupe(v)
    v, _ = s3_consumption(v)
    summ, _ = s4_summary(v)
    return summ


def deliver(records, cls):
    """Gives each row an arrival period ; the late-arriving row arrives one period later."""
    corrupted, n = corrupt(records, cls)
    data = [dict(k, arrival=k["period"]) for k in corrupted]
    if cls == "late_arrival":
        data += [dict(k, arrival="2026-08") for k in records
                 if k["period"] == "2026-07" and k["subscriber"] <= "A040"]
    return data, n


STAGES_BATCH = [a for a, _ in STAGES]
STAGES_STREAM = ["watermark"] + STAGES_BATCH + ["state_keys", "state_index"]


def batch(data, baseline):
    """Run , processes ALL rows that have ARRIVED up to the end of the previous period from the beginning."""
    rows_read, detector, stage, period = 0, None, None, None
    for i, p in enumerate(PERIODS):
        arrived = [k for k in data if PERIODS.index(k["arrival"]) < i]
        rows_read += len(arrived)
        v = arrived
        for step in range(6):
            if step > 0 and v:
                v, _ = STAGES[step - 1][1](v)
            for cname, cstage, cf in CHECKS:
                if cstage == step and detector is None and v and cf(v, baseline):
                    detector, stage, period = cname, step, p
    return {"rows_read": rows_read, "detector": detector, "stage": stage, "period": period,
            "state": 0, "late_dropped": 0, "cells": full_pipeline(data)}


def stream(data, baseline):
    """Period window: a row is processed in the period it arrives , state is carried to the next period."""
    seen, last_index, cells = set(), {}, {}
    rows_read, late_dropped, state = 0, 0, 0
    detector, stage, period = None, None, None
    for p in PERIODS:
        window = [k for k in data if k["arrival"] == p]
        rows_read += len(window)
        v = [k for k in window if k["period"] == p]         # watermark: the window closed
        late_dropped += len(window) - len(v)
        for step in range(5):
            if step == 1:
                v, _ = s1_screen(v)
            elif step == 2:
                fresh, new_keys = [], set()
                for k in v:
                    key = (k["subscriber"], k["period"])
                    if key not in seen and key not in new_keys:
                        new_keys.add(key)
                        fresh.append(k)
                seen |= new_keys                            # state: seen key
                v = fresh
            elif step == 3:
                c = []
                for k in v:
                    try:
                        c.append(dict(k, m3=int(k["index"]) - int(last_index[k["subscriber"]])))
                    except (KeyError, TypeError, ValueError):
                        pass
                for k in v:
                    try:
                        last_index[k["subscriber"]] = int(k["index"])   # state: last index
                    except (TypeError, ValueError):
                        pass
                v = c
            elif step == 4:
                totals = {}
                for k in v:
                    totals[(k["region"], p)] = totals.get((k["region"], p), 0) + k["m3"]
                cells.update(totals)
                v = totals
            for cname, cstage, cf in CHECKS:
                if cstage == step and detector is None and v and cf(v, baseline):
                    detector, stage, period = cname, step, p
        state = max(state, len(seen) + len(last_index))
    return {"rows_read": rows_read, "detector": detector, "stage": stage, "period": period,
            "state": state, "late_dropped": late_dropped, "cells": cells}


raw = generate_source()
BASELINE = full_pipeline(raw)
intact = [dict(k, arrival=k["period"]) for k in raw]
sb, ss = batch(intact, BASELINE), stream(intact, BASELINE)
print("baseline summary", len(BASELINE), "cells , serving", s5_serve(BASELINE)[0])
print("mode   stages     rows read  carried state  detector")
for name, r, n in (("batch", sb, len(STAGES_BATCH)), ("stream", ss, len(STAGES_STREAM))):
    print(f"{name:<6} {n:>6} {r['rows_read']:>13} {r['state']:>14}  {r['detector']}")

baseline summary 55 cells , serving {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55}
mode   stages     rows read  carried state  detector
batch       5         13200              0  None
stream      8          2400           2600  None

With the intact run both modes produce the same 55 summary cells and the same five served values, and no check fires in either. The cost row is read here. Batch mode reads 13200 rows in twelve runs, because every run processes the accumulated history from the beginning; stream reads each row once, 2400 rows. In return, stream raises the pipeline’s stages from 5 to 8 (the watermark, and the storing of the seen key and of the last index) and carries up to 2600 entries: 2400 keys and 200 subscriber indexes. The state carried by batch mode is zero.

Detection Period

The block below runs the six corruption classes in both modes and writes in which period and by which check each one is caught.

outcome = {}
for cls in CORRUPTIONS:
    data, n = deliver(raw, cls)
    outcome[cls] = (n, batch(data, BASELINE), stream(data, BASELINE))


def label(r):
    if r["detector"] is None:
        return "not caught"
    return f"{r['detector']}/{r['period']}/{PERIODS.index(r['period']) - PERIODS.index('2026-07')}"


def dirty(h):
    return sum(1 for key in set(h) | set(BASELINE) if h.get(key) != BASELINE.get(key))


print("corruption         corrupted  batch: check/period/lag      stream: check/period/lag")
for cls, (n, b, s) in outcome.items():
    print(f"{cls:<18} {n:>9}  {label(b):<28} {label(s)}")

print("\ncorruption          dirty cells/55 serving changed    in stream")
print("                     batch  stream    batch stream dropped rows")
for cls, (n, b, s) in outcome.items():
    e = ["no", "yes"]
    print(f"{cls:<18} {dirty(b['cells']):>7} {dirty(s['cells']):>7} "
          f"{e[s5_serve(b['cells'])[0] != s5_serve(BASELINE)[0]]:>8}"
          f"{e[s5_serve(s['cells'])[0] != s5_serve(BASELINE)[0]]:>7} {s['late_dropped']:>12}")

corruption         corrupted  batch: check/period/lag      stream: check/period/lag
late_arrival              40  distribution/2026-08/1       distribution/2026-07/0
duplicated                60  uniqueness/2026-08/1         uniqueness/2026-07/0
schema_changed           200  schema/2026-08/1             schema/2026-07/0
unit_changed             200  distribution/2026-08/1       distribution/2026-07/0
missing_partition         46  not caught                   partition/2026-07/0
type_drift               200  type/2026-08/1               type/2026-07/0

corruption          dirty cells/55 serving changed    in stream
                     batch  stream    batch stream dropped rows
late_arrival             0      10       no     no           40
duplicated               0       0       no     no            0
schema_changed          10      10      yes    yes            0
unit_changed            10      10       no     no            0
missing_partition        2       2      yes    yes            0
type_drift               0       0       no     no            0

Looking One Period Earlier

The first table gives the trade-off in a single row: in all five classes that are caught, batch mode says 2026-08 and stream mode says 2026-07. The period lag is 1 in batch mode and 0 in stream. The stage lags are exactly the same: schema, type and uniqueness are caught at stage 0, and distribution at stage 4. The mode does not change at which stage the corruption is caught; it changes only in which period one looks. Freshness here is not a serving property but a check property.

What a period means shows in the output: in batch mode the corrupted period has been served once by the time it is caught: 2026-07 has closed, the summary has been written, the served value has been read. This difference cannot be closed with a tolerance setting, because the lag of batch mode comes not from the check but from the schedule.

The second table shows what is independent of the mode. The duplicated row, type drift and unit change do not change the five served values in either mode; the first two do not even leave a dirty cell, the unit change dirties ten cells but the serving looks flawless. The only thing that sees these three is a check.

The Window’s Fix of the Check

The fifth row is the most unexpected result of the table. The missing partition is never caught in batch mode and is caught at stage 0 in stream mode, and moreover the same check code runs in both modes. The partition check looks at the region set and does not see the period; in the batch run all twelve periods are in front of it, the set is complete because north shows up in the other eleven periods, and the check stays green. In the stream run the set put in front of the same check is the window of a single period; there north is absent and the check fires.

The correctness of a check comes not only from its code but also from the set put in front of it. Although this looks like an advantage of stream mode, it is a side effect; the way to get the same result in batch mode is not to change the mode but to write the check period-aware.

The Cost of State

The first row of the second table reverses the trade-off. When the forty late-arriving rows arrive one period later, batch mode processes the accumulated history from the beginning and backfills the rows: all 55 summary cells stay correct. Stream drops the same rows at the watermark and, since the window has closed, never looks at them again. The result is 10 dirty summary cells, that is, 0.1818 of the cells.

The mechanism is the carried state. Because the 2026-07 readings of forty subscribers were never processed, the last-index state stays at 2026-06; when 2026-08 arrives, the consumption is computed from a two-period difference. In batch mode such a row drops, because the previous period is not in the table; in stream it is computed wrongly, because the state holds an old value. A missing row is a countable loss, not a wrong value.

The last column is the real warning of this section. Despite ten dirty cells, the five served values are exactly the same in both modes: what falls short in July is in excess in August, and the average of twelve periods does not move. Stream mode had caught this corruption in 2026-07; had it not, it could not have been found by looking at any output.

Summary

Batch mode processes the accumulated history from the beginning after a period closes, stream processes a row in the period in which it arrives; both use the same five stages and the same five checks, and their only difference is the window.
In all five classes that are caught, the period lag of batch mode is 1 and that of stream is 0; the stage lags are exactly the same. The mode changes not at which stage but in which period the corruption is seen.
The cost of stream is the number of stages rising from 5 to 8 and the carrying of 2600 entries; in return, the rows read fall from 13200 to 2400.
The period window fixes the flawed partition check: the same code never catches the missing partition in batch mode and catches it at stage 0 in stream mode.
For the late-arriving row the trade-off reverses: batch mode leaves all 55 cells correct, stream drops 40 rows and dirties 10 cells; even so, the five served values are the same in both modes.

Next Step

The pipeline was built and counted over five lessons: its stages were determined, the rows each stage drops were written, where the roles stand was separated, the data was broken down into source classes and the schedule it would run on was measured in two modes. One thing was never asked: whether the row the pipeline carries is correct. Everything counted is where a row came from, where it goes and when it is looked at; not a single measure that looks at the row itself was written. The checks were found ready-made, but where they come from was not defined. The next topic does this: it turns the quality dimensions into measurable checks, counts the detection lag of each check and measures the check’s own flaw.

To keep your progress and take notes, Log in