Windowed Aggregations and Joins in ksqlDB

Introduction

Stream processing rarely operates on unbounded data without temporal boundaries. When you need running totals, patterns within a time interval, or correlations between events that occurred near each other, windowed operations become essential. ksqlDB exposes these through a declarative SQL interface that compiles to Kafka Streams topologies, handling state management and time semantics for you.

The windowing model builds directly on Stream Processing fundamentals. Every windowed operation forces explicit decisions about window type, size, grace period, and timestamp extraction. Those decisions have direct operational consequences: they determine how much state your query accumulates, how long it retains that state, and how it handles out-of-order events that arrive after a window has nominally closed.

Because ksqlDB translates each statement into a Kafka Streams topology backed by persistent state stores, every windowed aggregation or join creates one or more RocksDB state stores, each backed by a changelog topic in Kafka. Their size and retention drive your cluster’s disk usage, memory pressure, and recovery time during rebalances.

Window Types and Their Operational Characteristics

ksqlDB supports three window types: tumbling, hopping, and session. (Kafka Streams added a separate sliding-window type in KIP-450, but ksqlDB does not expose a WINDOW SLIDING syntax; if you need fine-grained overlap, approximate it with a hopping window.) The choice affects both query semantics and resource footprint. The underlying mechanics are documented in the Kafka Streams DSL windowing reference and in the ksqlDB time-and-windows guide.

Tumbling Windows

Tumbling windows are fixed-size, non-overlapping, gap-less intervals. With a 5-minute tumbling window, each event belongs to exactly one window based on its timestamp. They are the simplest to reason about and the most resource-efficient, because state for a window can be discarded once the window closes and its grace period expires.

CREATE TABLE hourly_order_totals AS
SELECT
    product_id,
    COUNT(*) AS order_count,
    SUM(quantity) AS total_quantity,
    SUM(price * quantity) AS total_revenue
FROM orders_stream
WINDOW TUMBLING (SIZE 1 HOUR)
GROUP BY product_id
EMIT CHANGES;

Short windows produce more state-store segments, each smaller; longer windows reduce segment count but increase the memory held per segment. Track state-store size through the Kafka Streams state-store metrics (covered below) and watch RocksDB memory through kafka.streams:type=stream-state-metrics.

Hopping Windows

Hopping windows are fixed-size, overlapping intervals defined by a window size plus an advance interval. They are useful when you need updates more frequently than the window size implies, such as a 24-hour moving average that advances every hour.

CREATE TABLE rolling_daily_metrics AS
SELECT
    sensor_id,
    AVG(temperature) AS avg_temp,
    MAX(temperature) AS max_temp,
    MIN(temperature) AS min_temp
FROM sensor_readings
WINDOW HOPPING (SIZE 24 HOURS, ADVANCE BY 1 HOUR)
GROUP BY sensor_id
EMIT CHANGES;

The operational cost scales with the overlap factor: a 24-hour window advancing by 1 hour means each event participates in 24 windows simultaneously, multiplying state-store write amplification and aggregation CPU. ksqlDB has no first-class server properties for RocksDB write buffers or compression; to tune RocksDB you supply a class via ksql.streams.rocksdb.config.setter (see State Store Management below).

Session Windows

Session windows group events separated by a period of inactivity. Unlike tumbling and hopping windows, their size is data-driven: you specify a maximum inactivity gap, and a session closes when no events arrive for that duration. For a focused walkthrough, see How to Implement Session Windows for User Activity Tracking.

CREATE TABLE user_sessions AS
SELECT
    user_id,
    COLLECT_LIST(page_url) AS page_sequence,
    COUNT(*) AS page_views,
    (MAX(ROWTIME) - MIN(ROWTIME)) / 1000 AS session_duration_seconds
FROM clickstream
WINDOW SESSION (30 MINUTES)
GROUP BY user_id
EMIT CHANGES;

Session windows have a distinct operational hazard: a session for a continuously active key keeps extending, so its state grows until activity finally stops. Bound how long ksqlDB retains old (closed) windows with the RETENTION clause inside the window spec — for example WINDOW SESSION (30 MINUTES, RETENTION 2 HOURS, GRACE PERIOD 10 MINUTES). RETENTION must be at least window size plus grace period. (Under the hood this drives the Kafka Streams windowstore.changelog.additional.retention.ms setting; there is no ksql.streams.state.store.retention.ms property.) Session stores also compact more often than fixed windows because of the merge logic that combines adjacent session fragments.

Stream-Stream Joins with Windowing

Stream-stream joins in ksqlDB are always windowed: two streams cannot be joined without a temporal boundary, because join state would otherwise grow unbounded. ksqlDB enforces this by requiring a WITHIN clause.

CREATE STREAM enriched_orders AS
SELECT
    o.order_id,
    o.product_id,
    o.quantity,
    p.price,
    o.quantity * p.price AS line_total,
    o.order_time
FROM orders_stream o
INNER JOIN product_price_updates p
    WITHIN 1 HOUR
    ON o.product_id = p.product_id
EMIT CHANGES;

WITHIN 1 HOUR defines a symmetric window: an order joins any price update whose timestamp falls within one hour before or after it. (Use the WITHIN (<before>, <after>) form for an asymmetric window.) Both sides accumulate state — orders waiting for a matching price update and price updates waiting for a matching order — so total state is roughly proportional to event rate times window duration on each side. For high-throughput streams with wide windows this consumes substantial disk and page cache. Watch fetch behavior through the consumer MBean kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*; fetch latency tends to rise once state stores outgrow available page cache.

Join Grace Periods and Late Arrivals

The WITHIN clause accepts an optional GRACE PERIOD. If you omit it, ksqlDB applies a default grace period of 24 hours, which keeps join state open far longer than most production windows require — a frequent cause of oversized state stores. Set it explicitly to the lateness you actually expect. The Handling Late-Arriving Events with Allowed Lateness in ksqlDB article goes deeper.

CREATE STREAM enriched_orders_grace AS
SELECT
    o.order_id,
    o.product_id,
    o.quantity,
    p.price,
    o.quantity * p.price AS line_total
FROM orders_stream o
INNER JOIN product_price_updates p
    WITHIN 1 HOUR GRACE PERIOD 15 MINUTES
    ON o.product_id = p.product_id
EMIT CHANGES;

The grace period (GRACE PERIOD follows WITHIN, not inside its parentheses) controls how long after a window’s end records may still arrive and join; later records are dropped. A 1-hour window with a 15-minute grace period keeps each window’s state alive 15 minutes past its end. For LEFT/OUTER joins, the grace period also gates when non-matched rows are emitted, avoiding the spurious early results you get with a zero grace period. Place state stores on dedicated volumes with sufficient IOPS using ksql.streams.state.dir.

Timestamp Semantics and Event Time Processing

Windowed operations depend on correct timestamps. ksqlDB processes on event time: by default it uses the Kafka record timestamp (ROWTIME), and you can override that with a column from the value. Avoid relying on wall-clock processing time, which makes results non-deterministic and unreproducible.

CREATE STREAM orders_stream_with_ts (
    order_id VARCHAR KEY,
    product_id VARCHAR,
    quantity INT,
    price DOUBLE,
    order_time VARCHAR
) WITH (
    KAFKA_TOPIC = 'orders',
    VALUE_FORMAT = 'JSON',
    TIMESTAMP = 'order_time',
    TIMESTAMP_FORMAT = 'yyyy-MM-dd HH:mm:ss'
);

When TIMESTAMP is set to a VARCHAR column you must supply a TIMESTAMP_FORMAT parseable by Java’s DateTimeFormatter; if the column is BIGINT or TIMESTAMP, omit TIMESTAMP_FORMAT. Without an explicit TIMESTAMP, ksqlDB uses the record’s Kafka timestamp metadata, which the broker stamps as either CreateTime (producer-assigned) or LogAppendTime (broker-assigned). LogAppendTime reflects when the broker received the message, not when the event occurred, and will skew your windows. See the ksqlDB time-and-windows guide. For consistent results, configure producers (or topics) for CreateTime and declare the timestamp column explicitly.

Custom Timestamp Extraction

ksqlDB has no inline timestamp-extractor hook in a CREATE STREAM source definition, but you can normalize timestamps in a derived stream (optionally with a UDF) and then override TIMESTAMP on the result.

CREATE STREAM normalized_orders AS
SELECT
    order_id,
    product_id,
    quantity,
    price,
    STRINGTOTIMESTAMP(order_time_string, 'yyyy-MM-dd HH:mm:ss') AS order_epoch
FROM raw_orders_stream
EMIT CHANGES;

CREATE STREAM orders_with_ts
WITH (KAFKA_TOPIC='orders_with_ts', VALUE_FORMAT='JSON', TIMESTAMP='order_epoch')
AS
SELECT * FROM normalized_orders
EMIT CHANGES;

STRINGTOTIMESTAMP returns a BIGINT epoch (milliseconds since the Unix epoch), so the downstream WITH (... TIMESTAMP='order_epoch') override needs no TIMESTAMP_FORMAT. (On newer ksqlDB versions, PARSE_TIMESTAMP plus the TIMESTAMP data type is the recommended replacement for the older STRINGTOTIMESTAMP UDF.) The two-step approach adds a small latency but gives you full control and an operational checkpoint: parsing failures surface in the intermediate topic and the processing log instead of silently corrupting windowed state.

State Store Management and Monitoring

Every windowed query creates persistent RocksDB state stores. These are the primary operational concern in production: they consume disk, hold memory for block cache, and must be restored during rebalances or restarts.

Disk Sizing and Configuration

State-store disk usage is a function of event rate, key cardinality, window size, and grace period. A rough estimate per stateful task:

disk_bytes ≈ keys_in_window × (avg_key_size + avg_value_size) × overlap_factor × 2

The factor of 2 accounts for RocksDB write amplification and compaction overhead; overlap_factor is 1 for tumbling windows and equals size÷advance for hopping windows. Size for peak active key cardinality within the retained window span, not raw event rate — duplicate keys collapse into one entry per window. Note that the local on-disk store is not multiplied by the changelog’s replication factor; replication applies to the changelog topic in the brokers, which you size separately.

Put state stores on a dedicated path to avoid contention with broker log directories:

ksql.streams.state.dir=/data/ksqldb/state

To tune RocksDB beyond the defaults, set ksql.streams.rocksdb.config.setter to a class you provide that implements org.apache.kafka.streams.state.RocksDBConfigSetter. That class is where you control block cache size, write buffers, and compression — there are no individual ksql.streams.rocksdb.* properties for those knobs.

Monitoring State Store Health

Kafka Streams exposes per-store metrics under kafka.streams:type=stream-state-metrics (the store-scope tag is rocksdb-state-id, rocksdb-window-state-id, or rocksdb-session-state-id depending on store type), including put-rate, put-latency-avg, and fetch-latency-avg. High put latency usually points to insufficient disk IOPS or compaction pressure; high fetch/get latency suggests the block cache is too small for the working set. RocksDB memory usage is also surfaced through ksqlDB’s RocksDB metrics group. Build alerts on the metrics your JMX exporter actually emits, for example:

groups:
  - name: ksqldb_state_stores
    rules:
      - alert: StateStorePutLatencyHigh
        expr: kafka_streams_stream_state_metrics_put_latency_avg > 1e7  # 10 ms in ns
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "State store put latency high (IOPS or compaction pressure)"

      - alert: RecordsDroppedAsLate
        expr: rate(kafka_streams_stream_task_metrics_dropped_records_total[5m]) > 0
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Records dropped as late past the grace period"

(Exact exporter metric names depend on your JMX-to-Prometheus mapping; the underlying MBeans are stream-state-metrics and stream-task-metrics.)

Changelog Topic Management

Each state store is backed by a changelog topic named from the service ID and query ID, for example _confluent-ksql-default_query_CTAS_HOURLY_ORDER_TOTALS_0-Aggregate-Aggregate-Materialize-changelog. For windowed stores, ksqlDB sizes the changelog’s retention.ms from the window’s RETENTION (plus grace), so set retention on the query rather than on the topic:

SET 'ksql.streams.cache.max.bytes.buffering' = '10485760'; -- 10 MB record cache
-- then create the windowed query with an explicit RETENTION clause, e.g.
-- WINDOW TUMBLING (SIZE 1 HOUR, RETENTION 7 DAYS, GRACE PERIOD 15 MINUTES)

ksql.streams.cache.max.bytes.buffering controls the Kafka Streams record cache (default 10 MB) and governs how aggressively EMIT CHANGES results are deduplicated before being forwarded — a larger cache means fewer, more consolidated downstream updates. Keep RETENTION at or above window size plus grace, and ensure the changelog can hold the full retained span; if a store has to restore from a truncated changelog, the missing windows are lost.

Operational Patterns for Production Deployments

Scaling and Partitioning

Windowed state is partitioned by the grouping key. Uneven key distribution causes state skew, where some ksqlDB nodes carry disproportionately large stores. Favor high-cardinality grouping keys and watch per-task state metrics.

CREATE TABLE user_activity_windows AS
SELECT
    user_id,
    COUNT(*) AS event_count
FROM clickstream
WINDOW TUMBLING (SIZE 5 MINUTES)
GROUP BY user_id
EMIT CHANGES;

Avoid grouping high-throughput streams by low-cardinality attributes like status or event_type. If you must, add a salt to spread state across more partitions:

CREATE TABLE salted_aggregations AS
SELECT
    CONCAT(event_type, '_', CAST(CAST(RANDOM() * 10 AS INT) AS VARCHAR)) AS salted_key,
    COUNT(*) AS event_count
FROM events_stream
WINDOW TUMBLING (SIZE 1 MINUTE)
GROUP BY CONCAT(event_type, '_', CAST(CAST(RANDOM() * 10 AS INT) AS VARCHAR))
EMIT CHANGES;

RANDOM() returns a DOUBLE in [0.0, 1.0), so cast to INT before stringifying to get clean salt buckets. This distributes state at the cost of a downstream merge step to reconstruct the unsalted aggregate.

Graceful Shutdown and State Preservation

When stopping a ksqlDB server for maintenance, shut it down cleanly so state stores flush and you avoid full changelog restoration on restart. The bundled ksql-server-stop script sends a TERM signal and waits for the process to exit:

ksql-server-stop

ksql-server-stop takes no timeout flag; if you need a hard deadline, wrap it with the OS timeout command. ksqlDB does not expose a ksql.server.state metric. To confirm health and progress, watch the liveness-indicator metric (constant 1 while the server is emitting metrics) and the per-query query-status metric, which mirrors the underlying Kafka Streams application state.

Multi-Datacenter Considerations

Windowed operations across datacenters are sensitive to clock skew, which can assign events to the wrong window. Keep clocks tight with NTP and prefer event-time timestamps stamped at the source. For active-active topologies, partition by a datacenter identifier so events from different sites do not interleave within a partition:

CREATE STREAM partitioned_orders AS
SELECT
    dc_id,
    order_id,
    product_id,
    quantity,
    price,
    order_time
FROM orders_stream
PARTITION BY dc_id
EMIT CHANGES;

Troubleshooting Common Windowed Operation Failures

State Store Memory Pressure

If RocksDB block cache is undersized for the working set, or a single partition receives an outsized share of keys, you will see heavy GC, off-heap growth, or OutOfMemoryError. RocksDB allocates most of its memory off-heap (block cache, write buffers), so size the container memory limit to include it, not just the JVM heap. Cap and bound RocksDB memory through a RocksDBConfigSetter (set via ksql.streams.rocksdb.config.setter), and reduce window size, grace period, or per-key cardinality to limit accumulation.

To inspect or reset the state of a specific query’s Kafka Streams application, use the application reset tool with the query’s application.id. Always start with --dry-run, which prints the actions without performing them:

kafka-streams-application-reset \
    --bootstrap-servers localhost:9092 \
    --application-id _confluent-ksql-default_query_CTAS_HOURLY_ORDER_TOTALS_0 \
    --input-topics orders \
    --dry-run

This resets offsets and deletes internal topics for a clean replay; it does not report store sizes — for sizing, read the state-store metrics above or measure the on-disk ksql.streams.state.dir.

Late-Arriving Data Gaps

Gaps in windowed output usually mean events are arriving past the grace period and being dropped. Late drops are recorded by the Kafka Streams task metric dropped-records-total / dropped-records-rate under kafka.streams:type=stream-task-metrics; the processing log also captures detail. If the drop rate is non-zero, widen the grace period or fix upstream latency. For emitting only final results once a window closes (rather than on every update), KIP-328 suppression complements grace-period tuning; in ksqlDB this surfaces as EMIT FINAL for windowed aggregations (which requires ksql.suppress.enabled=true).

Join Spooling and Disk Pressure

Wide join windows drive heavy disk I/O as RocksDB spills to disk. Symptoms include rising disk utilization on the state volume and growing query latency. Narrow the join window or pre-filter the streams to cut cardinality before the join:

CREATE STREAM high_value_orders AS
SELECT * FROM orders_stream
WHERE quantity * price > 1000
EMIT CHANGES;

CREATE STREAM enriched_high_value AS
SELECT o.*, p.price
FROM high_value_orders o
INNER JOIN product_price_updates p
    WITHIN 30 MINUTES
    ON o.product_id = p.product_id
EMIT CHANGES;

Fewer events into the join means proportionally smaller state stores and less disk pressure.

Query Lag and Consumer Group Health

A windowed query that falls behind its input shows rising consumer lag. Monitor records-lag-max on the consumer MBean kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*. Persistent lag means the query cannot keep up with arrivals. Scale out by adding ksqlDB servers or by raising stream threads:

ksql.streams.num.stream.threads=4

Each thread handles a subset of partitions, so more threads add parallelism but also more concurrent state stores — ensure the node has the memory and CPU headroom before increasing the count.

Conclusion

Windowed aggregations and joins give ksqlDB powerful temporal abstractions, but they demand operational discipline. Window type, grace period, timestamp source, and state-store configuration directly determine reliability and cost in production. Ground your tuning in the real Kafka Streams mechanics — RETENTION and GRACE PERIOD on the window, a RocksDBConfigSetter for RocksDB, and the stream-state-metrics/stream-task-metrics MBeans for visibility — and you can run windowed ksqlDB queries at scale with predictable behavior.