Streaming, Event-Time Windows & Watermarks

Batch processing waits for a bounded dataset — yesterday's file — then processes it all at once. Simple, but results are only as fresh as the last batch.

Streaming processes an unbounded sequence of events continuously as they arrive: market agents reporting prices from their phones, sensors, clicks. Results update within seconds.

Streaming forces three questions batch never asks: Which time do I group by? When is a result final? What do I do with data that arrives late or twice?

Every event has two timestamps:

• Event time — when the thing actually happened (the price was observed at 08:00:05).
• Processing time — when our system received it (the phone had no signal; it arrived at 08:00:19).

If you group by processing time, delayed events land in the wrong window and your "08:00–08:10 average price" quietly includes prices from other periods. Correct analytics groups by event time.

Tumbling windows are fixed, non-overlapping buckets: for 10-second windows, an event at t falls in [⌊t/10⌋·10, ⌊t/10⌋·10 + 10). An event at 1,238.4 s goes to [1230, 1240).

With out-of-order data you can never be sure no more events for a window will arrive. A watermark is the system's assumption about how late events can be:

W(t) = max event time seen so far − L, where L is the allowed lateness.

• Any window whose end ≤ W(t) is finalised and emitted — its result won't change.
• An event with event_ts < W(t) is late: its window is already closed. Route it to a side output (late_events.jsonl) for auditing or correction instead of silently dropping it.

Choosing L is a trade-off: a small L gives fast, final results but more late events; a large L catches stragglers but holds results back longer and keeps more state in memory.

Mobile clients retry when they don't get an acknowledgement, so the same event can arrive twice. Counting both would inflate volumes and skew averages.

Our Lab 9 consumer keeps state: a set of seen_ids. The order of checks matters:

watermark = max_event_ts - allowed_lateness
if event_id in seen_ids:        route("duplicates")      # 1. drop retries first
elif event_ts < watermark:      route("late_events.jsonl") # 2. too late for its window
else:
    windows[(int(event_ts) // 10) * 10].add(event)        # 3. bucket by event time
    seen_ids.add(event_id)

State can't grow forever: once a window is final, the ids of its events can be expired. Accounting identity for a run: emitted = windowed + duplicates + late.