Streaming Window Processing

Streaming window processing aggregates movement telemetry as it arrives, grouping observations by the time they happened rather than the time they were received, and deciding explicitly how long to wait for stragglers before declaring a window final.

Everything difficult about it follows from one fact: mobility devices deliver late, and they deliver late in ways that correlate with exactly the situations the analysis cares about. A vehicle in an underground car park, a phone in a tunnel, a tracker in a rural notspot — each produces a burst of delayed events that arrive long after the window they belong to would have closed on wall-clock time. A streaming pipeline that ignores this is not faster than a batch one; it is a batch pipeline that silently drops the data hardest to collect.

Prerequisites

• A synchronized, monotone timeline. Every window boundary is meaningless without it; see time-series synchronization strategies for the UTC normalization and clock-drift work this assumes.
• Provenance on the timestamp. The stream must record which clock produced the event time — sensor, device OS or gateway — because they differ by up to thirty seconds and the watermark is sized against that difference.
• A partitioning key. Usually the entity id, so that state and ordering are per device rather than global.
• Runtime. The concepts below are runtime-agnostic; the examples use plain Python so the semantics stay visible, and map directly onto Flink, Kafka Streams, Spark Structured Streaming or Beam.

Event Time, Processing Time, and the Gap Between Them

The practical consequence is that “how late is late?” is a fleet question rather than an engineering one. A single watermark across mixed populations either drops most of the slow population’s data or holds the fast population’s results hostage to it. The usual resolution is to partition the stream by population and give each its own allowed lateness, accepting that different partitions finalise at different times and that the consumer must handle that.

Window Shapes and What Each Costs

The state column is the one that decides architecture. A 60-minute sliding window advancing every minute holds sixty accumulators per entity simultaneously; across 50 000 vehicles with a 200-byte accumulator that is roughly 600 MB before any framework overhead, and it grows linearly with fleet size. Incremental aggregations — those that can be updated with a single number rather than by retaining the members — are what keep that tractable, which is why mean and count are cheap in a sliding window and exact median is not.

The Watermark Decision

Implementation Walkthrough

The state machine below is deliberately framework-free: it shows the semantics that a runtime implements for you, which is exactly what you need to reason about when the runtime does something surprising.

from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime, timedelta


@dataclass
class WindowState:
    """Incremental accumulator — never retains members, so state stays O(1)."""
    count: int = 0
    dist_m: float = 0.0
    emitted: bool = False

    def add(self, dist_m: float) -> None:
        self.count += 1
        self.dist_m += dist_m


@dataclass
class TumblingWindower:
    """
    Event-time tumbling windows with a watermark and bounded allowed lateness.

    Emits (entity, window_start, count, dist_m, revision) tuples. Every emission
    is keyed by (entity, window_start), so a downstream upsert makes replay safe.
    """
    width: timedelta
    allowed_lateness: timedelta
    state: dict = field(default_factory=lambda: defaultdict(WindowState))
    watermark: datetime | None = None

    def _window_start(self, t: datetime) -> datetime:
        # Anchor to the epoch so bin edges are identical across every worker
        # and across every re-run. Never anchor to "first event seen".
        epoch = datetime(1970, 1, 1, tzinfo=t.tzinfo)
        n = (t - epoch) // self.width
        return epoch + n * self.width

    def push(self, entity: str, event_time: datetime, dist_m: float) -> list[tuple]:
        """Ingest one event; return any window results that became final."""
        if event_time.tzinfo is None:
            raise ValueError("event_time must be timezone-aware (UTC).")

        # The watermark is the highest event time seen, minus allowed lateness.
        if self.watermark is None or event_time > self.watermark + self.allowed_lateness:
            self.watermark = event_time - self.allowed_lateness

        w_start = self._window_start(event_time)
        w_end = w_start + self.width

        if w_end + self.allowed_lateness <= self.watermark:
            # Beyond the seal: count it, do not silently discard it.
            return [("__dropped__", w_start, entity, dist_m, -1)]

        st = self.state[(entity, w_start)]
        st.add(dist_m)

        out = []
        # Close every window whose end is now behind the watermark.
        for (ent, start), s in list(self.state.items()):
            if start + self.width <= self.watermark and not s.emitted:
                s.emitted = True
                out.append((ent, start, s.count, s.dist_m, 0))
            elif start + self.width <= self.watermark and s.emitted and (ent, start) == (entity, w_start):
                # A late-but-allowed event revised an already-emitted window.
                out.append((ent, start, s.count, s.dist_m, 1))
        return out

    def evict(self) -> int:
        """Drop sealed windows. Call periodically or state grows without bound."""
        if self.watermark is None:
            return 0
        cutoff = self.watermark - self.allowed_lateness
        stale = [k for k, _ in self.state.items() if k[1] + self.width <= cutoff]
        for k in stale:
            del self.state[k]
        return len(stale)

Three properties of that implementation are the ones worth carrying into any runtime. Window starts are anchored to the epoch rather than to the first event, so two workers processing different partitions agree on the bin edges and a re-run reproduces them. Emissions carry a revision number, so a consumer can tell a first result from a correction. And evict exists at all — a windowed stream without eviction is a memory leak with a schedule.

Calibration and Parameter Tuning

Of these, the idle-partition timeout is the one that surprises people. A watermark advances with the minimum across partitions, so a single quiet device or an empty Kafka partition holds every window open indefinitely. Every serious runtime offers an idleness timeout; it needs to be set, not left at its default.

Integration and Compatibility

The strongest requirement on a streaming windower is that it agrees with the batch pipeline computing the same thing. That agreement is not automatic, and it usually breaks in one of three places: bin edges anchored differently, a stop-gap threshold that differs between the session window and trajectory segmentation, or a batch job that sees late data the stream had already sealed.

The workable pattern is to make the batch job authoritative and the stream provisional. The stream emits fast, revisable results keyed by window and entity; the batch job recomputes the same keys from the archive once the data is complete and upserts over them. Consumers read one table and get low latency now and correctness later, without either pipeline having to pretend it is the other. That also gives a free correctness check — the size and sign of the batch corrections is a direct measurement of whether the watermark is right.

Downstream, rolling statistics for mobility metrics and dynamic time binning strategies both assume windows that mean the same thing every time they are computed, which is precisely what epoch-anchored edges and idempotent emission provide.

Validation and Testing Patterns

Streaming pipelines fail in ways that batch pipelines cannot, and the tests have to target those specifically rather than checking that the aggregation arithmetic is right.

Replay determinism. Feed the same recorded event sequence twice and assert the emitted results are byte-identical. Anything that breaks this — wall-clock defaults, unordered map iteration, a window start anchored to the first event seen — is a bug that will show up as unreproducible numbers long before anyone suspects the windower.

Out-of-order injection. Take a clean stream, shuffle a controlled fraction of events backwards in time by a known amount, and assert the final totals match the in-order run. This is the only test that actually exercises the watermark, and it is the one most often missing. Vary the shuffle distance across the allowed-lateness boundary so both the update path and the drop path are covered.

Idle-partition simulation. Stop feeding one partition and assert that windows on the others still close. A watermark held back by a silent partition is the most common cause of a stream that “stops producing output” while every health check reports green.

State-size assertion. Track open windows per entity and total state bytes as a metric, and alert on growth. A missing eviction is invisible until the job dies, and it always dies at the least convenient moment — the point of highest traffic, which is also the point of highest state.

In This Section

• Handling late-arriving GPS events with watermarks — measuring the lag distribution and turning it into an allowed-lateness setting.
• Computing tumbling and sliding windows over telemetry streams — the state cost of each shape and how to keep sliding windows incremental.

FAQ

What is the difference between event time and processing time?

Event time is when the observation happened on the device; processing time is when it reached the pipeline. For mobility telemetry the gap is routinely minutes and occasionally hours, because vehicles park in garages, phones sleep and networks drop. Windowing on processing time is fast and produces numbers that change whenever the network does; windowing on event time is slower and produces numbers that mean what they say.

How long should the watermark allow for late data?

Measure, do not guess. Collect arrival minus event time over a representative week and set allowed lateness at a high percentile, typically the 99th. For urban scooter fleets that is often under a minute; for HGVs parking in underground bays it can exceed four hours. Then decide explicitly what happens beyond it — dropped and counted, or routed to a correction path.

Do I need exactly-once processing for movement aggregates?

You need idempotent output, which is easier and usually sufficient. If each result is keyed by window plus entity and written as an upsert, a replay overwrites rather than adds, and the aggregate is correct however many times the record was processed. True exactly-once delivery is only necessary when the sink cannot be keyed.

How much state does a windowed stream hold?

Roughly active entities × open windows per entity × per-window state. A 60-minute sliding window advancing every minute keeps 60 windows open per entity, so 50 000 vehicles with a 200-byte accumulator is about 600 MB before overheads. This is why sliding windows are expensive and why incremental aggregations matter.

Should session windows or fixed windows define a trip?

Session windows, because a trip is defined by inactivity rather than by the clock. A session window closes after a configurable gap of no events, which is exactly the trip-splitting rule used in batch segmentation. Keeping the same gap threshold in both paths is what allows live and next-day trip counts to agree.

Related

• Time-Series Synchronization Strategies — the UTC and clock-provenance work every window boundary depends on.
• Rolling Statistics for Mobility Metrics — the batch counterpart of the sliding-window aggregations here.
• Dynamic Time-Binning Strategies — choosing the width the stream will then apply.
• Gap Filling in Sparse Trajectories — what to do about the intervals the stream never receives.
• Trajectory Segmentation — the batch definition a session window has to match.

Back to Temporal Aggregation & Window Mapping