Stream Processing — Watermarks, Windows, Exactly-Once

Why this session exists

A loop reading Kafka is not yet a correct stateful processor. This lesson defines time, window membership, progress, emission, state cleanup and sink effects separately. It targets Spark 3.5.7 and Flink 1.18 where their contracts differ. It teaches tumbling, sliding, session and custom windows, stream joins, enrichment and recovery—not only a minimal counting example.

Outcomes: distinguish event/processing time; define watermark, lateness and output policy; choose windows and an engine from requirements; trace state/offset/sink recovery; and implement replay-safe aggregate replacement with a negative control showing why additive UPSERT is unsafe.

Prerequisites

Batch versus streaming establishes completeness and freshness; Kafka establishes replay, transactions and offsets. Kafka is a common source, not a requirement for exactly-once state recovery.

Postmarks versus delivery times

A letter's postmark and arrival time answer different questions. Likewise, event time is a source timestamp and processing time is the operator clock when it handles the record. Ingestion time is another recorded boundary. Event-to-processing delay includes source/network/backlog delay and clock error; a large gap is not automatically a broken producer.

An event-time sales report attributes a late 10:00 event to the 10:00 bucket. An operational monitor asking how many requests the worker handled at 10:15 correctly uses processing time. Replaying historical data into processing-time windows changes results because the question changed; that is not universally a bug.

The original diagram incorrectly placed a mobile-network outage after Kafka had already ingested the event; this version makes the delayed consumer boundary explicit.

Historical context

The 2015 Dataflow model separates event-time windows, triggers and refinement; Beam exposes a model spanning bounded and unbounded data. Spark's older DStreams API and Structured Streaming are not identical, and Flink's record-at-a-time APIs offer different state/timer controls. Avoid unsupported “first engine with proper semantics” claims: compare the versioned APIs needed by the workload, not a supposed universal latency hierarchy.

Four window types with exact membership

• Tumbling: fixed nonoverlapping intervals, e.g. [0,60), [60,120). Event 60 belongs to the second, not both. The origin/timezone is part of the definition.
• Sliding: size 60, slide 20 gives overlapping windows. Event 45 belongs to [0,60), [20,80), [40,100) for this aligned origin. More membership can increase state/output; duplicate counting across windows is intentional, not a global total.
• Session: key-specific intervals merge when activity falls within an inactivity gap. Events at 0 and 18 with gap 10 initially form separate intervals; a late event at 9 connects them. A sink holding the two earlier results must replace/retract them, not just append a third result.
• Custom/global: a count trigger can emit every N records, but a global window does not automatically bound retained state. Define trigger, accumulation, eviction and recovery semantics. Count windows are useful outside large organizations too.

Session boundary equality differs with API conventions; the exact model below uses intervals [event,event+gap) and merges overlapping, not merely touching, intervals. It is an explicitly specified algorithm model, not a promise to reproduce every engine's boundary implementation.

Watermarks, firing and cleanup are different decisions

A watermark is event-time progress under a policy. It is not proof that older records cannot arrive and not simply wall-clock now minus a delay.

Spark 3.5.7: call withWatermark on the event-time column before supported stateful aggregation. Progress depends on maximum observed event time and a delay threshold; processing/batch timing also matters. The guarantee is one-sided: data delayed less than the threshold is not dropped by watermarking; older data may or may not be processed. State cleanup additionally requires a compatible aggregation and output mode. Update mode emits changed aggregate rows; append mode for supported watermarked aggregates waits until finalization; complete mode emits the whole result and cannot drop old aggregate state merely to bound it. A 30s trigger is scheduling intent, not a 30s latency guarantee [Spark streaming].

Flink 1.18: a default event-time window trigger fires as watermark progress passes the window's firing timestamp. Allowed lateness retains eligible window state after the initial firing; late accepted records can cause further results before cleanup. A side output can retain too-late records for correction. Zero default allowed lateness is not zero out-of-orderness: watermark generation can already lag observed events. Triggers and custom policies can change when output occurs [Flink windows].

For Flink multiple input channels, downstream event-time progress is constrained by active input watermarks; idleness detection avoids one idle input blocking everything but accepts a late-data tradeoff when it resumes. Spark multiple-watermark policy defaults to a conservative minimum across input watermarks; choosing maximum can advance faster and drop slower-input data. Spark's global observed-time aggregation must not be described as identical to per-partition Flink watermark generation [Flink watermarks; Spark streaming].

Controlled lateness exercise: compare separate fresh checkpoints with delays 120s and 30s. Establish observed event-time progress first, then send timestamps targeting old windows. A record “90 seconds before wall-clock now” alone predicts nothing: if it is the first record, it may be the newest event time the query knows. Inspect actual progress, state rows and numRowsDroppedByWatermark; do not promise tenfold drops or subtract the configured delay twice. Choose retention from observed lag distribution, incident/backfill tails and acceptable correction loss, not P99 alone as a correctness guarantee.

Why joins need state—and which joins

A join must have both rows to emit a matched pair. A bounded hash join also uses state; batch is not “stateless.” With two unbounded append streams, a row may match future rows, so retaining all potentially useful rows can grow without bound. A time-range condition plus progress on both inputs can establish when no accepted future match is possible and allow eviction. A named tumbling window is not the only bound: interval joins, finite key lifecycles, changelog/upsert semantics or explicit TTL contracts can change requirements.

A stream-stream inner join without time bounds can be offered with unbounded state; it is not syntactically impossible. Even matched rows may need retention if future rows can produce more matches. Outer-join unmatched outputs need progress sufficient to conclude no on-time match remains. Adding an arbitrary TTL trades completeness for bounded state and must be visible in the contract.

A stream-table lookup avoids buffering two arbitrary append streams, but the table can still grow, be stale or require versioned history. “Current dimension” enrichment can change when replayed later. To reproduce historical results, embed the relevant value at event creation or perform a versioned temporal lookup using both business-effective and availability semantics where needed.

Enrichment tradeoffs and the standing-query model

External lookup: avoids replicating a large reference table in job state, but adds latency, rate limits and availability dependencies. It is only as current/consistent as that service, cache and read policy. Async/batching can help but complicate ordering and retries.

Broadcast reference state: makes bounded reference data local to tasks; replicated size, update ordering, restart loading and consistency determine feasibility. It does not mean throughput is limited only by CPU or every event sees the newest reference value.

Enrich at producer: captures context with the event and simplifies downstream work, but changes producer responsibility/payload and may still require a transactional snapshot. Decide which version is correct before optimizing the lookup.

A standing query keeps state as events flow through it. Treat state as an operated database: size, checkpoint, restore, evolve and clean it up. A long checkpoint interval can increase replay work, but catch-up also depends on available throughput and source retention. Session duration can remain unbounded under continuous activity. A compacted changelog can represent a current table, but deletion/retention means it need not preserve all historical versions.

Exactly-once effect boundaries

Recovery needs replayable input positions, consistent operator state and a sink protocol that tolerates repeat attempts. A transactional Kafka source additionally requires read-committed visibility if aborted records must be hidden, but every input producer does not need a Kafka transaction. Flink checkpointed state and a supported transactional sink can coordinate external publication; Spark foreachBatch is at-least-once by default [Flink fault tolerance; Spark streaming].

Do not conflate repeated network delivery with duplicate application effects. A framework can replay computation and still expose one committed effect under a defined protocol. It cannot roll back arbitrary emails or payment calls. Receiver-supported idempotency keys, transactional outboxes or cooperative transactions are needed. There is no unqualified theorem here that every meaningful “exactly-once delivery” API is impossible, nor a claim that idempotence is always cheaper.

For aggregate outputs, replacing n with a deterministic full count can be idempotent; adding n on every retry is not. Also reject stale versions: an old batch's replacement arriving after a new one can regress the result. A batch ledger should include query/checkpoint lineage, batch ID and enough payload identity to reject conflicting reuse. Resetting a checkpoint requires a new lineage and a deliberate reconciliation—not reusing batch zero against old state.

Hands-on: windows, session merging and sink recovery

This complete CPU/SQLite program validates window membership against an independent enumeration, demonstrates late session bridging, and tests transactional sink replacement, stale-batch rejection, conflicting reuse and crash-before/after-commit. It models a serial query lineage; it is not a Spark/Flink execution trace.

from collections import Counter
import sqlite3, json, hashlib
 
def windows(t,size,slide):
    if size<=0 or slide<=0: raise ValueError('positive durations required')
    last=(t//slide)*slide
    return [(s,s+size) for s in range(last-size+1,last+1) if s%slide==0 and s<=t<s+size]
for t in range(-60,121):
    oracle=[(s,s+60) for s in range(-120,181,20) if s<=t<s+60]
    assert windows(t,60,20)==oracle
assert windows(60,60,60)==[(60,120)]
assert windows(45,60,20)==[(0,60),(20,80),(40,100)]
 
def sessions(times,gap):
    out=[]
    for t in sorted(times):
        if out and t<out[-1][1]:
            out[-1]=(out[-1][0],max(out[-1][1],t+gap),out[-1][2]+1)
        else: out.append((t,t+gap,1))
    return out
assert sessions([0,18],10)==[(0,10,1),(18,28,1)]
assert sessions([0,18,9],10)==[(0,28,3)]
assert sessions([0,10],10)==[(0,10,1),(10,20,1)]
 
c=sqlite3.connect(':memory:')
c.executescript('''CREATE TABLE batches(lineage TEXT,batch_id INT,digest TEXT,PRIMARY KEY(lineage,batch_id));
CREATE TABLE results(lineage TEXT,user_id TEXT,start INT,end INT,n INT,batch_id INT,
 PRIMARY KEY(lineage,user_id,start));''')
def publish(lineage,batch_id,rows,fail=False):
    if len({(u,s) for u,s,e,n in rows})!=len(rows): raise ValueError('duplicate result identity')
    if any(e<=s or n<0 for u,s,e,n in rows): raise ValueError('invalid aggregate')
    digest=hashlib.sha256(json.dumps(sorted(rows)).encode()).hexdigest()
    with c:
        prior=c.execute('SELECT digest FROM batches WHERE lineage=? AND batch_id=?',(lineage,batch_id)).fetchone()
        if prior:
            if prior[0]!=digest: raise ValueError('batch identity conflict')
            return
        for user,start,end,n in rows:
            c.execute('''INSERT INTO results VALUES(?,?,?,?,?,?)
             ON CONFLICT(lineage,user_id,start) DO UPDATE SET end=excluded.end,n=excluded.n,batch_id=excluded.batch_id
             WHERE excluded.batch_id>results.batch_id''',(lineage,user,start,end,n,batch_id))
        c.execute('INSERT INTO batches VALUES(?,?,?)',(lineage,batch_id,digest))
        if fail: raise RuntimeError('before commit')
rows=[('u1',0,60,2)]
try: publish('q1',0,rows,True)
except RuntimeError: pass
assert c.execute('SELECT count(*) FROM results').fetchone()[0]==0
publish('q1',0,rows);publish('q1',0,rows) # crash after commit, then retry
assert c.execute('SELECT n FROM results').fetchone()[0]==2
publish('q1',2,[('u1',0,60,4)])
publish('q1',1,[('u1',0,60,3)]) # delayed older replacement must not regress
assert c.execute('SELECT n FROM results').fetchone()[0]==4
try: publish('q1',0,[('u1',0,60,9)])
except ValueError: pass
else: raise AssertionError('conflict accepted')
# Negative control: additive UPSERT-style repetition is not idempotent.
wrong=0
for _ in range(2): wrong+=2
assert wrong==4
assert len(sessions([0,18,9],10)) != len(sessions([0,18],10))
print({'window_oracle_cases':181,'sliding_45':windows(45,60,20),'bridged_session':sessions([0,18,9],10),'latest_count':4,'duplicate_addition':wrong})
c.close()

Spark-to-PostgreSQL reference integration

This Spark 3.5.7/psycopg2 reference uses a bounded driver-side sink for a teaching fixture, not scalable production output. Supply the matching Spark Kafka connector (spark-sql-kafka-0-10_2.12:3.5.7 for the matching Scala build), a precreated table and a durable unique checkpoint. A libpq service named stream_lesson supplies connection settings through operator-managed configuration; no password belongs in the lesson. No Kafka, PostgreSQL or Spark service was run here.

Precreate, in the authorized test database:

CREATE TABLE click_agg (
  lineage text NOT NULL, window_start timestamptz NOT NULL,
  window_end timestamptz NOT NULL, user_id text NOT NULL,
  n bigint NOT NULL CHECK(n>=0), batch_id bigint NOT NULL,
  PRIMARY KEY(lineage,window_start,user_id)
);

import os
from datetime import datetime, timezone
from contextlib import closing
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import StructType, StringType, TimestampType
import psycopg2
 
spark=(SparkSession.builder.appName('window_reference')
 .config('spark.sql.session.timeZone','UTC').getOrCreate())
schema=StructType().add('user_id',StringType()).add('ts',TimestampType())
lineage=os.environ['LESSON_QUERY_LINEAGE']
checkpoint=os.environ['LESSON_CHECKPOINT_URI']
raw=(spark.readStream.format('kafka').option('kafka.bootstrap.servers','127.0.0.1:9092')
 .option('subscribe','lesson-clicks').option('startingOffsets','earliest')
 .option('kafka.isolation.level','read_committed').load())
parsed=raw.select(F.from_json(F.col('value').cast('string'),schema).alias('e')).select('e.*')
# Explicit fail-fast policy rather than silently filtering invalid records.
valid=parsed.selectExpr("CASE WHEN user_id IS NULL OR ts IS NULL THEN raise_error('invalid event') ELSE user_id END AS user_id",'ts')
agg=(valid.withWatermark('ts','2 minutes').groupBy(F.window('ts','1 minute'),'user_id')
 .count().select(F.col('window.start').alias('window_start'),F.col('window.end').alias('window_end'),'user_id',F.col('count').alias('n')))
 
def write_batch(df,batch_id):
    # Collect epoch seconds, not naive driver-local Python timestamps.
    rows=(df.select(F.col('window_start').cast('long').alias('start_epoch'),
      F.col('window_end').cast('long').alias('end_epoch'),'user_id','n')
      .limit(1001).collect())
    if len(rows)>1000: raise ValueError('teaching sink row bound exceeded')
    with closing(psycopg2.connect(service='stream_lesson', options='-c timezone=UTC')) as conn:
        with conn,conn.cursor() as cur:
            for r in rows:
                cur.execute('''INSERT INTO click_agg VALUES(%s,%s,%s,%s,%s,%s)
                  ON CONFLICT(lineage,window_start,user_id) DO UPDATE
                  SET n=EXCLUDED.n,window_end=EXCLUDED.window_end,batch_id=EXCLUDED.batch_id
                  WHERE EXCLUDED.batch_id>=click_agg.batch_id''',
                  (lineage,datetime.fromtimestamp(r.start_epoch,timezone.utc),
                   datetime.fromtimestamp(r.end_epoch,timezone.utc),r.user_id,r.n,batch_id))
try:
    query=(agg.writeStream.outputMode('update').foreachBatch(write_batch)
      .option('checkpointLocation',checkpoint).trigger(processingTime='30 seconds').start())
    query.awaitTermination()
finally:
    spark.stop()

What each block does and what it does not promise

The timestamp column drives one-minute tumbling membership. This reference uses classic PySpark 3.5.7, not Spark Connect. At the Python boundary, TimestampType.fromInternal uses driver-local datetime.fromtimestamp; setting only the Spark SQL timezone does not set the Python process timezone. The sink therefore collects integer epoch seconds and constructs timezone-aware UTC datetimes for psycopg2. The minute-aligned windows here have whole-second boundaries; this conversion is not suitable for preserving arbitrary subsecond timestamps without a microsecond-aware variant. The watermark is before aggregation. Update output supplies full changed counts, not deltas. foreachBatch can retry; the sink replaces values and refuses older batch IDs. Under one active query, stable lineage/checkpoint and deterministic same-batch rows, retrying a committed batch has the same result. The stronger CPU model additionally detects conflicting same-batch payloads; this reference relies on the framework's deterministic same-batch contract and must not be shared with concurrent/reset writers. A durable checkpoint is not a /tmp directory silently lost on restart.

Collect is deliberately capped at 1,001 rows to fail before publishing oversized output; it bounds driver results, not upstream compute. A production-scale sink needs a distributed staging/commit protocol with unique query/batch identities, deterministic conflict handling and recovery of abandoned staging. Merely overwriting a shared staging table and issuing MERGE is not a generic solution: concurrent attempts, duplicate JDBC task writes and stale batches need design.

This count is over source records, not deduplicated business events; producer/application duplicates require a separate event-ID contract. Invalid records stop the query, so a production system needs a governed quarantine/repair path rather than silently dropping them. Sessions that merge or aggregates that delete rows require retractions/replacements for obsolete result keys; this tumbling-count sink does not implement that different output contract.

Crash exercise answer: fail before transaction commit and no result should appear; fail after sink commit but before Spark records batch completion and the retry should replace with the same count. Verify both state/offset recovery and sink output. The executed SQLite test proves its local transaction model; it does not establish PostgreSQL/Spark integration behavior.

Session replacement is more than a value UPSERT

The session model produces two intervals for 0 and 18, then one interval after 9 bridges them. Updating the row keyed by start=0 alone leaves the obsolete start=18 row in the sink and double-counts a visit. The following bounded SQLite companion replaces all sessions for one user in a complete finite snapshot, atomically. Its scope is explicitly different from Spark's changed-row update output: deleting every user row for a partial delta would lose valid sessions.

import sqlite3
c = sqlite3.connect(':memory:')
c.execute('CREATE TABLE sessions(user TEXT,start INT,end INT,n INT,PRIMARY KEY(user,start))')
def replace_complete_user(user, rows, fail=False):
    with c:
        c.execute('DELETE FROM sessions WHERE user=?', (user,))
        c.executemany('INSERT INTO sessions VALUES(?,?,?,?)', [(user,*r) for r in rows])
        if fail: raise RuntimeError('rollback replacement')
def current():
    return c.execute('SELECT start,end,n FROM sessions WHERE user=? ORDER BY start', ('u',)).fetchall()
old = [(0,10,1), (18,28,1)]
merged = [(0,28,3)]
replace_complete_user('u', old)
try: replace_complete_user('u', merged, True)
except RuntimeError: pass
assert current() == old
replace_complete_user('u', merged)
replace_complete_user('u', merged)
assert current() == merged
print({'session_rows_after_replayed_complete_replacement':current()})
c.close()

A concurrent or delayed snapshot writer additionally needs lineage/version fencing, as in the earlier batch-version discussion. This companion intentionally has one serial writer and does not implement those controls. A changelog alternative publishes explicit deletes for obsolete keys and upserts for the merged key in one commit boundary.

Worked oral answers: processing-time replay can move yesterday's sales into today's processing bucket; event-time membership preserves the intended historical question. A watermark is policy-defined progress, with a delay knob distinct from Flink allowed lateness and from sink emission. Choose a cooperating transaction protocol when output records and consumed offsets must be atomic; choose receiver-enforced idempotence when repeat delivery is acceptable but repeat effects are not. Neither choice covers a receiver that honors no such protocol.

Production reality and engine selection

Slow-source scenario: if desktop events advance progress while mobile is delayed, aggressive global progress can discard mobile records. Conservative multi-input policy, idleness decisions, separate pipelines or a correction path trade freshness/state/completeness. There is no verified Netflix five-minute incident in this lesson; the counterexample is sufficient to teach the issue.

Replay-to-audit scenario: emitting twice after a checkpoint failure can inflate a downstream append-only audit. Transactional Kafka output with appropriately configured checkpoint integration and read-committed consumers, or an idempotent external sink, can address the declared boundary. Do not attach an invented Uber loss or 500ms penalty to this mechanism.

Default-lateness scenario: no Spark watermark generally means no watermark-based eviction, not a universal default zero that drops every out-of-order event. Flink zero allowed lateness is distinct from its watermark delay. Monitor late/drop counts and compare reconciled outputs to retained source snapshots when the business requires completeness. Reconciliation is a useful control, not a theorem that streaming without a batch job is unauditable.

Choose between Flink and Spark on at least three concrete dimensions: deadline under representative load and recovery, required operators/state/timers and source/sink contracts, and deployment/team ecosystem. Flink's DataStream timers/CEP/broadcast-state tools may fit event-driven state machines; Spark integrates relational transformations with an existing Spark batch estate. Both have version/connector limits. Spark continuous processing has different guarantees/operator support and does not support foreachBatch; it is not a drop-in low-latency switch for this aggregation. No universal sub-100ms versus 500ms benchmark is asserted.

Connections to the surrounding curriculum

Airflow can orchestrate bounded reconciliation over streaming-populated tables; dbt can transform compatible snapshots; quality checks inspect completeness and late data rather than only schema. CDC requires consistent change interpretation. Online features need event-time/availability-aware windows so training and serving do not diverge. Fraud enrichment combines these time and state choices with an action deadline.

Recall and worked answers

1. Event versus processing time? Source occurrence timestamp versus operator handling clock; the correct choice depends on the question.
2. Watermark assertion and older records? Progress under a policy; older records may update retained state, be dropped or enter a correction path according to engine/operator rules.
3. Window selection? Per-minute click count: tumbling; inactivity-defined visit: session; one-hour dashboard every five minutes: sliding; every N events: count/custom trigger with explicit retention.
4. Exactly-once pieces? Replayable input, consistent processing state/position recovery, and a cooperating sink/effect protocol—not universally a transactional Kafka producer.
5. Is processing time wrong? Wrong for reproducing source-time sales buckets under replay; right for operational work-received-now metrics.
6. Effectively-once? Repeated attempts yield one intended effect through idempotent/transactional design; an additive UPSERT is not that design.
7. Watermark delay? Use observed lag plus tail/incident/backfill requirements, state/finality budget and a correction policy; P99 leaves a tail and is not a completeness proof.
8. Flink versus Spark? Compare deadline, operators/state/sink contract and operations on representative workloads, not product latency slogans.
9. Why conservative per-input progress? Fast input alone cannot prove slow input complete. Minimum active progress can delay output; idleness/maximum policies change the loss risk. No unverified company attribution is required.

Failure answer: x=x+batch_count repeats effects even inside UPSERT. Atomically dedup batch identity or replace a deterministic result with stale-version protection; session merges also require removing obsolete keys.

Sources and next step

Checked 2026-10-07: Spark 3.5.7 streaming, Flink 1.18 windows, watermark generation, fault tolerance, Kafka 3.7 design. Exact passages/version/hash/entailment are retained separately.

Next: Airflow orchestration. The exact CPU/SQLite model ran; external integrations and actual cluster latency remain unexecuted, not incomplete conceptual coverage.

Original resource trail and reading purpose

The original optional pre-read destinations are retained below, not used as primary evidence for numerical claims, vendor incidents, or current API defaults. Video identity, duration, attribution and playback have not been independently verified here; search-result links are discovery links, not a particular verified lecture. Mutable latest documentation may differ from this lesson's pinned release. Use the versioned primary sources above for the governing contract.

• Foundations of streaming SQL by Tyler Akidau — original optional video/discovery destination; historical label, not verified attribution.
• Webinar: Deep Dive on Apache Flink State - Seth Wiesman — original optional video/discovery destination; historical label, not verified attribution.
• Spark Structured Streaming — Watermarks + Windows — original optional video/discovery destination; historical label, not verified attribution.
• Streaming 102 — original optional reading destination; historical label, not verified attribution.
• Flink — Timely Stream Processing — original optional reading destination; historical label, not verified attribution.
• Spark Structured Streaming — Window Ops on Event Time — original optional reading destination; historical label, not verified attribution.

Read with a task: separate time assignment, window membership, triggers and cleanup; compare Spark watermark/output modes with Flink 1.18 rather than importing one engine's terms into another.

Series navigation: previous lesson; learning-plan hub. The original series used a 90-minute intuition → visual → hands-on → production → recall structure; that historical pacing is not a measured completion time for this revised lesson.