Lab 1 — Build a watermarked tumbling-window aggregator

Objective

By the end of this lab you will have a runnable Python module, stream.py, that consumes a simulated event stream — deliberately out of order, with one genuinely late record — and produces correct per-window counts using a watermark to decide when each tumbling window is complete. The late record is detected and dropped on purpose, not silently miscounted. The module uses only the Python standard library, so it costs nothing and runs anywhere.

Skills exercised:

• Assigning events to tumbling windows by event time
• Advancing a monotonic watermark from the maximum event time seen
• Closing windows and releasing their state when the watermark passes their end
• Detecting and handling a late event per a stated policy

Prerequisites and setup

Prerequisites: the three lessons in this module — Event time vs. processing time, Windowing and watermarks, and Stream processing patterns. This lab turns the watermark bookkeeping from Lesson 2 and the late-event policy from Lesson 3 into working code.

Setup: you need Python 3.10 or newer. There are no packages to install; everything uses the standard library. Verify your version and create a working folder.

macOS / Linux

python3 --version
mkdir de205-stream-lab && cd de205-stream-lab

Windows (PowerShell)

python --version
mkdir de205-stream-lab; cd de205-stream-lab

Expected output from the version command is a line like Python 3.11.6. If the command is not found, install Python from python.org before continuing. In the steps below, use python3 on macOS/Linux and python on Windows.

note

Every step edits the single file stream.py in this folder. If you close your terminal and come back, re-run cd de205-stream-lab to return to it; your file is still there. Each step adds to the file, so keep the code from earlier steps in place.

Step 1 — Create the event source

Goal: build the generator that emits the stream, with events deliberately out of order and one that will arrive late.

Create stream.py with the constants and the generator:

from collections import defaultdict

WINDOW_SIZE = 10  # seconds spanned by each tumbling window
ALLOWED_LATENESS = 5  # how far the watermark trails the newest event time


def generate_events():
    """Yield (event_time, sensor_id) pairs, deliberately out of order.

    Event times are whole seconds. The event at event_time 7 arrives long
    after its window has closed — it is the late event you will handle later.
    """
    script = [
        (2, "sensor-a"),
        (5, "sensor-b"),
        (8, "sensor-a"),
        (13, "sensor-c"),
        (3, "sensor-a"),   # out of order, but still within allowed lateness
        (16, "sensor-b"),
        (25, "sensor-a"),  # jumps the watermark forward
        (7, "sensor-b"),   # late: window [0, 10) has already closed
        (22, "sensor-c"),
        (28, "sensor-a"),
    ]
    for event in script:
        yield event

Checkpoint: count the events the generator produces.

python3 -c "import stream; print(sum(1 for _ in stream.generate_events()))"

You should see 10. If you see a different number, compare your script list against the ten rows above — a missing comma silently merges two tuples.

Step 2 — Assign events to tumbling windows

Goal: map any event time to the start of the tumbling window it belongs to.

Add this function to stream.py:

def window_start(event_time):
    """Return the start second of the tumbling window an event falls in."""
    return (event_time // WINDOW_SIZE) * WINDOW_SIZE

Checkpoint: check three event times land in the right windows.

python3 -c "import stream; print(stream.window_start(7), stream.window_start(13), stream.window_start(28))"

You should see 0 10 20: event time 7 is in window [0, 10), 13 is in [10, 20), and 28 is in [20, 30). If every value is 0, WINDOW_SIZE is not set to 10; check the constant from Step 1.

Step 3 — Aggregate with a watermark that closes windows

Goal: build the processor that counts events per window, advances a watermark, and closes any window the watermark has passed. This version does not yet handle late events — you add that in Step 4.

Add this class to stream.py:

class TumblingWindowProcessor:
    """Aggregate a stream into tumbling windows, gated by a watermark."""

    def __init__(self, window_size=WINDOW_SIZE, allowed_lateness=ALLOWED_LATENESS):
        self.window_size = window_size
        self.allowed_lateness = allowed_lateness
        self.watermark = None  # None until the first event arrives
        self.open_counts = defaultdict(int)  # window_start -> running count
        self.closed_counts = {}  # window_start -> final count

    def _window_end(self, start):
        return start + self.window_size

    def process(self, event_time, key):
        """Route one event into its window, then advance the watermark."""
        start = window_start(event_time)
        self.open_counts[start] += 1
        candidate = event_time - self.allowed_lateness
        if self.watermark is None or candidate > self.watermark:
            self.watermark = candidate
        self._close_ready_windows()

    def _close_ready_windows(self):
        for start in sorted(self.open_counts):
            if self._window_end(start) <= self.watermark:
                self.closed_counts[start] = self.open_counts.pop(start)

Checkpoint: feed a short in-order sequence and inspect the state.

python3 -c "import stream; p = stream.TumblingWindowProcessor(); [p.process(t, 'k') for t in [2, 5, 8, 13, 16]]; print(dict(p.closed_counts), dict(p.open_counts), p.watermark)"

You should see {0: 3} {10: 2} 11: three events closed in window [0, 10), two still open in [10, 20), and the watermark at 11 (the max event time 16 minus the 5-second lateness). If closed_counts is empty, the watermark never reached a window end — confirm _close_ready_windows compares _window_end(start) <= self.watermark.

Step 4 — Detect and drop late events

Goal: give the processor a deliberate late-event policy. A record whose window has already closed must not reopen it or corrupt a neighbor; here the policy is to route it to a late_events list and drop it from the counts.

First, add one line to __init__ to hold late events. Place it after the closed_counts line:

        self.late_events = []  # events that arrived after their window closed

Then replace your process method with this version, which checks for lateness before counting:

    def process(self, event_time, key):
        """Route one event into its window, or set it aside if it is late."""
        start = window_start(event_time)
        end = self._window_end(start)
        if self.watermark is not None and end <= self.watermark:
            self.late_events.append((event_time, key))
            return
        self.open_counts[start] += 1
        candidate = event_time - self.allowed_lateness
        if self.watermark is None or candidate > self.watermark:
            self.watermark = candidate
        self._close_ready_windows()

Checkpoint: feed the whole stream through process and inspect the result.

python3 -c "import stream; p = stream.TumblingWindowProcessor(); [p.process(t, k) for t, k in stream.generate_events()]; print(dict(p.closed_counts), 'late=', p.late_events)"

You should see {0: 4, 10: 2} late= [(7, 'sensor-b')]: window [0, 10) closed with four events (including the out-of-order 3), window [10, 20) with two, and the single late record (7, 'sensor-b') set aside instead of miscounted. If late is empty, the lateness guard is not firing — confirm it runs before open_counts[start] += 1.

Step 5 — Flush open windows and run the stream

Goal: close any windows still open when the stream ends, then wire the whole processor into a runnable program.

Add the flush method to the class (same indentation as process):

    def flush(self):
        """End of stream: close every window still open, then report."""
        for start in sorted(self.open_counts):
            self.closed_counts[start] = self.open_counts.pop(start)
        return self.closed_counts

Then add the driver code at the end of the file, outside the class:

def run():
    processor = TumblingWindowProcessor()
    for event_time, key in generate_events():
        processor.process(event_time, key)
    windows = processor.flush()
    return windows, processor.late_events


if __name__ == "__main__":
    windows, late = run()
    for start in sorted(windows):
        end = start + WINDOW_SIZE
        print(f"window [{start}, {end}): {windows[start]} events")
    print(f"late events dropped: {len(late)}")

Checkpoint: run the whole module.

python3 stream.py

You should see exactly four lines:

window [0, 10): 4 events
window [10, 20): 2 events
window [20, 30): 3 events
late events dropped: 1

Window [20, 30) now shows its three events because flush closed it after the stream ended. If it is missing, run is not calling processor.flush() before returning.

Validation

Verify the whole build against the objective: correct per-window counts and the late event handled as specified. Run this validation one-liner, which reruns the stream and checks both the window counts and the exact late record:

python3 -c "import stream; w, late = stream.run(); ok = (w == {0: 4, 10: 2, 20: 3}) and late == [(7, 'sensor-b')]; print('PASSED — windows correct and late event handled' if ok else f'FAILED — got {w} late={late}')"

Checkpoint: the command prints PASSED — windows correct and late event handled. If it prints FAILED, compare the reported counts against {0: 4, 10: 2, 20: 3}: a wrong window [0, 10) count usually means the out-of-order event 3 was dropped or the late event 7 was counted; a missing window [20, 30) means flush did not run.

Teardown

Nothing is billable — this lab runs entirely on your machine. Keep the de205-stream-lab folder if you want to reuse stream.py as a reference for the Structured Streaming module, which runs this same watermark logic on real tooling; otherwise delete it with rm -rf de205-stream-lab (macOS/Linux) or Remove-Item -Recurse de205-stream-lab (Windows).

Troubleshooting

Symptom	Likely cause	Fix
ModuleNotFoundError: No module named 'stream'	Running from the wrong folder	cd into de205-stream-lab, where stream.py lives
Step 1 checkpoint prints a number other than 10	A tuple in script is malformed	Check each row has both values and a trailing comma
window_start returns 0 for every input	WINDOW_SIZE not set to 10	Confirm the WINDOW_SIZE = 10 constant from Step 1
late list is empty after Step 4	Lateness guard runs after counting	The if ... end <= self.watermark check must come before open_counts[start] += 1
Window [0, 10) shows 3 instead of 4	Out-of-order event 3 treated as late	With ALLOWED_LATENESS = 5, event 3 is still valid; check the guard uses end <= watermark, not event_time
Window [20, 30) missing from the output	flush not called at end of stream	run must call processor.flush() before returning
Validation prints FAILED	Any of the above	Re-run the Step 4 and Step 5 checkpoints to isolate which count is wrong