Build your own mini stream processor · Stage 4 of 8

Tumbling/sliding windows with triggers

Windows turn an unbounded keyed stream into finite aggregations. In event time, a window fires when the watermark passes its end; events that arrive after that (beyond an allowed-lateness grace) are late and go to a side output.

New API this stage

TumblingWindows(size)                # .assign(ts) -> [Window]
SlidingWindows(size, slide)          # .assign(ts) -> [Window] (ts may be in many)
Window(start, end)                   # end exclusive
keyed.window(assigner).aggregate(agg_fn, allowed_lateness=0) -> WindowResult
run_windows(WindowResult) -> Fired   # list of (key, Window, result); .late side output

This stage's task

TumblingWindows(size).assign(ts) → one window [start, start+size), start = ts - (ts % size).
SlidingWindows(size, slide).assign(ts) → every window containing ts (stepping start down by slide; windows may start before 0, matching Flink).
run_windows: bucket each in-time event's value into a pane keyed by (key, window). When the watermark passes a window's end, fire it — emit (key, Window, agg_fn(values)) — in (start, key) order, and clear it.
Drop events late beyond allowed_lateness to result.late.
After all watermarks, the fired result equals a batch grouping of in-time events into windows (the oracle).

Constraints

Event-time triggers only (watermark-driven), never processing time or sleeps.
Deterministic fire order. Earlier stages still pass. Suite < 60 s.

Run it

de start FLK-BYO/04
de test

Hints

Hint 1

TumblingWindows(size).assign(ts) returns the single window [start, start+size) where start = ts - (ts % size). SlidingWindows(size, slide).assign(ts) returns EVERY window [s, s+size) that contains ts, where s steps down by slide from the latest aligned start; windows may start before 0 (Flink does this).

Hint 2

run_windows drives the chain, bucketing each event's value into panes keyed by (key, window). A window FIRES when the watermark passes its end (window.end <= current_watermark): emit (key, Window, agg_fn(values)) and drop the pane. Fire ready windows in (window.start, key) order for determinism.

Hint 3

Lateness: an event is late if max(window ends it belongs to) + allowed_lateness <= current_watermark at arrival — route those to the .late side output instead of a pane. The streaming result after all watermarks must equal a batch grouping of in-time events (the oracle).

Run it locally

uv run de start FLK-BYO/04-windows-triggers
uv run de test