Skip to content

Window types

Three window shapes with durable RocksDB-backed state by default. Tumbling and sliding also support the opt-in resident storage mode. This page is the catalog: what each kind is and how to declare it. The mechanics that are the same everywhere — the aggregate you pass to value=, the on_close_each / on_close callbacks, early-fire, and the processing-/event-time model — live on the Windowing page. Failover, dynamic construction and sessions at scale are on Windowing — advanced.

Choose the boundaries: fixed (Tumbling, Sliding: you choose the size up front) vs data-driven (Session: boundaries come from activity).

Windows are built inside a class-based handler so the window object can register itself on self in __init__ and persist across batches.

The first example uses processing time. The event-time snippets below assume the input topic declares event_time="event_ts_ms"; windows inherit that clock by omitting time_column. For Kafka timestamps, use the Kafka clock configuration. A window cannot select a different clock from its subscription.

Tumbling

Fixed-size, non-overlapping windows: event-time timestamp t is assigned to the window starting at trunc(t / size) * size, where trunc means integer division toward zero, and ending at start + size. For example, t=-1 with size=1000 gives a start of 0, rather than -1000. The everyday choice for time-bucketed counts and aggregates.

PersistentTumbling

import pyarrow as pa
from turbine import KafkaBroker, RecordBatch, Turbine, TurbineState
from turbine import aggregates as agg
from turbine import windowing as win

kafka = KafkaBroker(bootstrap="localhost:9092")
app = Turbine(kafka)

@app.subscribe(kafka.topic("events"), output=kafka.topic("counts"))
class count_per_region:
    def __init__(self) -> None:
        self._win = win.PersistentTumbling(
            self,
            name="events_per_region",
            size_ms=60_000,                # 1-minute windows
            key=["region"],
            value=agg.Count(),
            on_close_each=self._emit,
        )

    def process(self, batch: RecordBatch, state: TurbineState) -> RecordBatch | None:
        self._win.update(batch, state)
        return None                        # emissions happen on window close

    def _emit(self, state: TurbineState, window: win.Window, value: float) -> RecordBatch:
        return pa.RecordBatch.from_pylist([{
            "region": window.group["region"],
            "count": float(value),
            "window_start_ms": window.start_ms,
            "window_end_ms": window.end_ms,
        }])

app.run()

Constructing the window in __init__ is the recommended pattern: it's wired up before any batch arrives, so windows that were in-flight at a previous crash fire correctly on the new instance even with no fresh traffic. (To defer construction until the first batch — when the window identity depends on runtime data — see Dynamic windows.)

It survives node failure: accumulator state and the fire-time timer are persisted and checkpointed, so a migrated partition restores from the latest snapshot. See Failover-resilient windows for the completeness boundary.

Sliding

Overlapping fixed-size windows with a shorter slide step. A 5-minute size with a 1-minute slide emits one window every minute, each covering the last 5 minutes — rolling dashboards, smoothed ratios. size_ms must be a positive multiple of slide_ms.

PersistentSliding

self._win = win.PersistentSliding(
    self,
    name="rolling_error_rate",
    size_ms=5 * 60_000,
    slide_ms=60_000,                 # N = size/slide = 5 overlapping windows
    key=["region"],
    value=agg.Multi(errs=agg.Sum("is_error"), n=agg.Count()),
    on_close_each=self._emit,
    allowed_lateness_ms=5_000,       # optional
)

PersistentSliding uses a paned pre-aggregation model: each event is merged into one slide_ms-sized pane regardless of how many overlapping windows would cover it, and a window emits by merging its N = size_ms // slide_ms panes at close. Ingestion stays O(1) per event even when many windows overlap, which matters when the aggregate is expensive (Multi, sketch-backed TDigestQuantile). Expired panes are GC'd automatically. Everything else — event/processing time, allowed_lateness_ms, the close-callback choice, restart-safety — matches PersistentTumbling.

Key values may contain |. Turbine escapes separators in group identities and decodes them when closing a window. Values such as "tenant|region" remain distinct from a composite key containing separate "tenant" and "region" values; no sanitizing or hashing is required before update.

A key column must not be named window_start_ms (or window_end_ms, on session windows), nor like one of the aggregate's internal state columns (e.g. x_count for Variance("x"), events_count for Count("events")): the table delivered to the close callbacks — and the partials the state lane commits — merge columns by name, and the window's own bound or state column would silently overwrite the key — the fixed window kinds refuse it with a ValueError at construction, for all fixed window kinds. The natural traps are a second-stage window over another window's output topic, whose schema is precisely keys + window_start_ms + outputs — key on the original key columns and let the bound flow as data — and keying on a column that happens to carry an aggregate's state-column name.

Session

Tumbling and sliding windows have boundaries you choose up front. A session window has none: its boundaries come from the data. Each key accumulates events into a session for as long as they keep arriving; once a key falls silent for longer than gap_ms, the session closes, and the next event opens a fresh one. This is the natural shape for "a burst of activity" — a user's click session, a device's reporting burst, a login-to-logout span.

Sessions take gap_ms (the inactivity gap) instead of size_ms.

Merging is the defining behaviour. Sessions are computed per key. An event within gap_ms of an existing session extends it; an event (or run of events) that lands between two separate sessions and closes the gap on both sides merges them into one. In event-time mode this also happens for out-of-order and late data: a late event can bridge two open sessions that looked distinct when they formed. The window emitted to your callback always spans [earliest event, latest event + gap), so window.end_ms is data-driven — never assume start + a fixed size. Once a session has closed (the watermark reached session_end + allowed_lateness_ms), a row whose candidate session would also be closed is dropped or routed as late data. A later row can open a new session if its deadline is still ahead of the frontier; keep allowed_lateness_ms wide enough to cover your out-of-order spread so genuine continuations merge instead of fragmenting.

Beyond gap_ms, one optional knob matters:

  • max_duration_ms — a hard cap on total session length. Without it, a key that never goes quiet (a stuck sensor, a bot) keeps a single session open forever. With it, the session is force-closed once it reaches start + max_duration_ms, and subsequent events start a fresh session. Must be >= gap_ms; leave it 0 (default) to disable.

PersistentSession

The durable session window: state lives in RocksDB, so sessions survive restart and partition handoff, and the close fires on a precise timer like the other persistent kinds.

Session aggregate calculations follow the common numeric aggregate rules. If a PyArrow kernel does not support half-float input, cast it to Float32 or Float64 before updating the window.

self._win = win.PersistentSession(
    self,
    name="user_sessions",
    gap_ms=30 * 60_000,              # 30 min of inactivity ends a session
    key=["user_id"],
    value=agg.Multi(clicks=agg.Count(), spend=agg.Sum("amount")),
    on_close_each=lambda state, window, v: pa.RecordBatch.from_pylist([{
        "user_id": window.group["user_id"],
        "clicks": v.clicks,
        "spend": v.spend,
        "session_start_ms": window.start_ms,
        "session_end_ms": window.end_ms,    # data-driven, not start + size
    }]),
    allowed_lateness_ms=60_000,
    max_duration_ms=4 * 3_600_000,   # optional hard cap
)

Prefer PersistentSession whenever session state matters across restarts (most sessions are long enough — minutes to hours — that losing them on a crash hurts). At very high key cardinality it runs slower — but doesn't fall over — see Sessions at scale.