S049 · Stream Processing — Watermarks, Windows, Exactly-Once

Time is the hard part of streaming.

Module M05: Data Engineering · Session 49 of 130 · Track: DE ⏱️

What you'll be able to do after this session

Explain Stream Processing to a friend in one minute, out loud, no notes.
Recognise it when you see it in real code / systems / papers.
Complete the hands-on exercise at the end without looking up help.

Prerequisites

If you can't answer these in 30 seconds each, redo the linked session first:

S048 — Kafka — Topics, Partitions, Consumer Groups

Pre-read (skim before the session)

🎥 Intuition (5–15 min) · Stream Processing System Design Explained | Kafka, Flink, Spark — end-to-end design of a real stream processor, all the moving parts named.
🎥 Deep dive (30–60 min) · Windowing and Watermarks in Flink | Flink with Java (Confluent) — the definitive walkthrough of the two hardest concepts.
🎥 Hands-on demo (10–20 min) · Kafka in Docker: Step-by-Step Guide for Producers & Consumers — practical baseline: you need Kafka running to do any real streaming demo.
📖 Canonical article · Confluent — Time and Watermarks in Flink — the mental model for event time vs processing time.
📖 Book chapter / tutorial · Google — The Dataflow Model (paper) — the paper that gave the world watermarks, windowing, and triggers. Foundational.
📖 Engineering blog · Netflix — Keystone Real-time Stream Processing Platform — how streaming actually looks at petabyte-per-day scale.

(a) Intuition · 5 min

The one-paragraph mental model. Why this concept exists, what problem it solves, and the analogy that makes it stick.

In one sentence: Time is the hard part of streaming.

You're on a train, watching messages arrive on your phone. Someone's phone was in a tunnel for 10 minutes — their "I'll be at the station at 3pm" message just landed at 3:07pm. Is that message "late" or "on time"? From the message's point of view (it was sent at 2:55pm), it's on time. From your point of view (it arrived at 3:07pm), it's late. Now imagine you're running a "what happened between 3pm and 3:05pm?" report and you get that message at 3:07pm — do you include it? These are the questions stream processing exists to answer, and the answer is subtle enough that it took the industry 15 years and a dedicated Google paper (the Dataflow Model) to settle.

The core distinctions: event time is when the event happened at the source. Processing time is when your stream engine sees it. In batch, these are basically the same; in streaming, they can differ by seconds (network jitter), minutes (a laptop was offline), or days (a phone was in airplane mode). A window is a slice of the stream you're computing over: "5-minute tumbling window," "24-hour sliding window every hour," "session window per user." A watermark is the engine's best guess of "we've probably seen all events with event-time ≤ X." When the watermark passes a window's close, the window fires and emits a result — but late data may still arrive.

The gotcha to carry forever: streaming makes you decide, explicitly, what "correct" means when time is fuzzy. Do you emit at watermark and never update? Emit early speculative results and update? Wait a long time for late data at the cost of latency? "Exactly-once" is not a single knob — it's a chain (source, engine state, sink) and each link must be idempotent for the whole thing to hold. Beginners write streaming code assuming events arrive in order, at processing time, and the world matches their mental model. It never does.

(b) Visual walkthrough · 15 min

Diagrams, worked examples, step-by-step visuals.

The three "kinds of time" you must always distinguish:

Window types, in one table:

Window	Shape	Example use
Tumbling	Fixed size, non-overlapping	"Sales per 1-min bucket"
Sliding	Fixed size, overlaps by a step	"Sales in the last 5 min, updated every 30 sec"
Session	Dynamic gap between events per key	"User's browsing session"
Global	One big window, custom triggers	Continuous aggregations with early emits

Worked example — 1-minute tumbling window, watermark = event_time − 10s:

Events (event_time → processing_time):
  A@10:00:03 → arrives 10:00:04
  B@10:00:07 → arrives 10:00:08
  C@10:00:15 → arrives 10:00:16
  X@10:00:05 → arrives 10:01:20   ← LATE! (arrived 75s after event)
  D@10:01:02 → arrives 10:01:03

The window [10:00:00, 10:01:00) normally fires when watermark reaches 10:01:00 (i.e., processing time ≈ 10:01:10). It emits {A, B, C} and closes. When X arrives at 10:01:20, it targets the closed window — you now choose:

Drop late data (simplest, some data loss).
Send to a side-output for a nightly reconciliation job.
Configure allowedLateness=5min — the window re-fires with {A, B, C, X} — but downstream sinks must be idempotent so {A,B,C} isn't double-counted.

Exactly-once — the chain:

Break any link and you get at-least-once + duplicates. Flink's answer: distributed checkpoints (a coordinated snapshot of all operator state + Kafka offsets), plus transactional sinks (Kafka, Iceberg, JDBC with 2PC) so state and output move forward atomically. Spark Structured Streaming does the equivalent with a write-ahead log per micro-batch. Kafka Streams uses transactions to commit offsets + output atomically.

(c) Hands-on · 20 min

Code you type and run. Not pseudo-code — real, runnable, copy-pasteable.

Simulate late events and a tumbling window in pure Python (no Flink install needed).

# stream_lab.py — a mini stream processor demonstrating event time, watermarks, and windows
import heapq, collections, time, random
 
WINDOW_SIZE     = 10           # seconds (tumbling)
ALLOWED_LATENESS = 5           # seconds — accept late events this old
WATERMARK_DELAY  = 3           # emit when processing time ≥ window_end + this
 
# Fake event stream. Each event: (event_time, key, value).
# Some events are late (event_time much less than "now").
def event_source(n=100):
    now = 1000
    for i in range(n):
        now += random.random() * 2                 # advance ~1s per tick
        # 20% of events are "late" by 4-8s
        if random.random() < 0.2:
            et = now - random.uniform(4, 8)
        else:
            et = now - random.uniform(0, 1)
        yield (et, random.choice(["A","B","C"]), 1)
        time.sleep(0.02)
 
# Per-key window state: {(window_start, key): count}
state = collections.defaultdict(int)
emitted_windows = set()
 
max_event_time = 0
for event_time, key, val in event_source():
    max_event_time = max(max_event_time, event_time)
    watermark = max_event_time - WATERMARK_DELAY
 
    win_start = int(event_time // WINDOW_SIZE) * WINDOW_SIZE
    win_end   = win_start + WINDOW_SIZE
 
    # accept only if window is still open (or within allowed lateness)
    if win_end + ALLOWED_LATENESS < watermark:
        print(f"  DROPPED late event et={event_time:.1f} key={key} (win closed)")
        continue
 
    state[(win_start, key)] += val
 
    # fire any windows whose end <= watermark and haven't been fired yet
    for (ws, k), count in list(state.items()):
        if ws + WINDOW_SIZE <= watermark and (ws, k) not in emitted_windows:
            print(f"WIN [{ws:.0f}-{ws+WINDOW_SIZE:.0f}) key={k} count={count}  (watermark={watermark:.1f})")
            emitted_windows.add((ws, k))

Run it:

python3 stream_lab.py

Observe (checklist):

Each window fires only after the watermark has advanced past window_end — that's the delay streaming pays for correctness.
Watermark = max event time seen − 3s — a heuristic. Real engines (Flink) let you plug in custom watermark strategies.
Some events with et far in the past get printed as DROPPED — they missed the window even including allowed lateness.
Even the "on-time" events can arrive in slightly different processing-time order — event time is what matters for the window.
Windows fire once, so downstream must not re-count them — this is why sinks must be idempotent.

Try this modification: re-emit updated results when late events arrive (change emitted_windows from "set of fired windows" to "map from window → last count," and always emit an update when count changes). You've just implemented Flink's "accumulating" trigger mode — and now understand why downstream sinks must be idempotent upserts, not blind inserts.

(d) Production reality · 10 min

How this shows up in real systems, gotchas, war stories, common bugs.

The most famous public streaming story: Alibaba processes ~4 billion events/sec at peak during Singles' Day (Nov 11) using Flink. Their engineers have published multiple papers on what breaks at that scale — mostly checkpointing overhead and RocksDB state size. Lesson: state is the hard part of streaming, not throughput. A stateless map-filter pipeline scales trivially. A stateful join or session-window pipeline has state proportional to your key cardinality × window size, and eventually every state store (RocksDB in Flink, HDFS/S3 in Spark SS) hits limits. Rule of thumb: keep window sizes bounded (session-based, or with hard TTLs), and shard state by a high-cardinality key.

Watermark tuning is where you lose sleep. Too aggressive (small delay) → too much late data dropped, wrong results. Too conservative (large delay) → dashboards are minutes behind reality, angry PMs. Uber's answer: dual-tier — an "early speculative" window that emits at low latency, plus a "final" window with generous lateness that overwrites the speculative result. Downstream stores must support this (upsert semantics). Confluent's ksqlDB handles this with EMIT CHANGES.

Exactly-once end-to-end is a bar most teams don't clear. In practice, at-least-once + idempotent sinks (dedup by an event ID or upsert by primary key) is far more robust and much cheaper. Netflix's Keystone explicitly chose at-least-once for its scale. Reserve true exactly-once for financial ledgers and billing systems, where the extra cost is justified.

Failure recovery is where streaming systems earn their salary. A Flink checkpoint failure that leaves state in an inconsistent state means restoring from an older checkpoint and replaying Kafka from that point. If Kafka's retention was too short, you cannot recover — permanent data loss. Set Kafka retention to at least 3-7 days for topics feeding stateful streaming jobs, and monitor checkpoint duration + size — a checkpoint that takes 4 minutes is a smell that your state is too big.

Finally: backpressure is a feature, not a bug. When your sink is slow, Flink slows down consumption from Kafka. This is desirable — it prevents runaway memory growth. Alert on it; don't fix it by adding buffers.

(e) Quiz + exercise · 10 min

Explain the difference between event time, processing time, and ingestion time. Which one should windowing use, and why?
What is a watermark and what trade-off do you make when you set its delay large vs small?
Compare tumbling, sliding, and session windows with one use case for each.
What are the three "links in the chain" you must control to achieve end-to-end exactly-once processing?
Why is idempotent output usually preferred over transactional exactly-once sinks in real production?

Stretch (connects to S048 — Kafka): you're building a "revenue in the last hour, updated every 30 seconds" dashboard on top of Kafka. Sketch the full architecture: partition key on the topic, windowing choice, watermark strategy, sink store, and what happens when a bug requires re-running the last 24 hours' worth of results.

Explain-out-loud test

If you can't teach these 3 things to a friend without notes, redo the session:

What is Stream Processing? (one sentence, no jargon)
When would you use it? (one concrete example)
When would you NOT use it? (one anti-pattern)

What comes next

Next session (S050): Orchestration — Airflow, DAGs, Retries, Backfills
Previous (S048): Kafka — Topics, Partitions, Consumer Groups
Hub: The 6-Month Learning Plan

Part of a 130-session evergreen learning series. Session structure: (a) intuition · (b) visual walkthrough · (c) hands-on · (d) production reality · (e) quiz + exercise. Duration: 60 minutes.