--- title: 'Batch and Stream Processing' source: 'https://academia.sh/en/courses/introduction-to-data-engineering/batch-and-stream-processing' course: 'Introduction to Data Engineering' language: en updated: '2026-10-21T05:03:00+00:00' license: 'CC BY-SA 4.0' objectives: - '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.' --- # 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. 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. ```python # 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. ```python 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**.