10.6.3 – Aggregations & Windows

Module: Section 10 – Kafka Ecosystem & Tools
Section: 10.6 – Querying & Processing with ksqlDB
Goal: Master aggregations and windowing in ksqlDB

Aggregation Functions

SELECT
  user_id,
  COUNT(*) AS click_count,
  SUM(amount) AS total_amount,
  AVG(amount) AS avg_amount,
  MIN(amount) AS min_amount,
  MAX(amount) AS max_amount
FROM user_clicks
GROUP BY user_id
EMIT CHANGES;

Tumbling Windows

Fixed-size, non-overlapping windows

SELECT
  user_id,
  COUNT(*) AS clicks_per_hour,
  WINDOWSTART AS window_start,
  WINDOWEND AS window_end
FROM user_clicks
WINDOW TUMBLING (SIZE 1 HOUR)
GROUP BY user_id
EMIT CHANGES;

Hopping Windows

Fixed-size, overlapping windows

SELECT
  user_id,
  COUNT(*) AS clicks
FROM user_clicks
WINDOW HOPPING (SIZE 1 HOUR, ADVANCE BY 15 MINUTES)
GROUP BY user_id
EMIT CHANGES;

Session Windows

Activity-based windows

SELECT
  user_id,
  COUNT(*) AS actions_in_session
FROM user_actions
WINDOW SESSION (30 MINUTES)
GROUP BY user_id
EMIT CHANGES;

Advanced Windowing Techniques

Session Windows with Inactivity Gaps

Session windows track activity sessions with dynamic window boundaries based on user inactivity:

CREATE TABLE user_sessions AS
SELECT 
  user_id,
  COUNT(*) AS event_count,
  COLLECT_LIST(event_type) AS events,
  WINDOWSTART AS session_start,
  WINDOWEND AS session_end,
  (WINDOWEND - WINDOWSTART) / 1000 AS session_duration_seconds
FROM user_events
WINDOW SESSION (30 MINUTES)
GROUP BY user_id
EMIT CHANGES;

Key Features:

• Window closes after 30 minutes of inactivity
• Automatically adapts to user behavior
• Each user can have multiple concurrent sessions
• Sessions merge if events arrive within gap period

Example Flow:

User clicks at:
10:00:00 → Session starts
10:05:00 → Session extends
10:15:00 → Session extends
[30 minute gap - no events]
10:45:00 → Session closes at 10:15:00
10:45:00 → New session starts

Use Cases:

• ✅ Clickstream analysis
• ✅ User engagement tracking
• ✅ Shopping cart abandonment detection
• ✅ Application usage patterns

Grace Periods for Late-Arriving Data

Handle out-of-order events in windowed aggregations:

SELECT 
  product_id, 
  SUM(quantity) AS total_sold,
  WINDOWSTART AS hour_start
FROM orders
WINDOW TUMBLING (SIZE 1 HOUR, GRACE PERIOD 15 MINUTES)
GROUP BY product_id
EMIT CHANGES;

How It Works:

1. Window closes at normal time (e.g., 11:00)
2. Grace period extends window state for 15 minutes
3. Late events arriving between 11:00-11:15 still update the window
4. After 11:15, window is finalized and late events ignored

Example Timeline:

Window: 10:00 - 11:00
Grace Period: 11:00 - 11:15

Event arrives at 10:55 with timestamp 10:30 → ✅ Accepted
Event arrives at 11:10 with timestamp 10:45 → ✅ Accepted (within grace)
Event arrives at 11:20 with timestamp 10:50 → ❌ Ignored (past grace)

Key Benefits:

• ✅ Accepts late-arriving data up to grace period past window close
• ✅ Maintains accurate aggregates despite network delays
• ✅ Configurable per query based on latency requirements
• ✅ Balances accuracy vs. result freshness

Configuration Examples:

-- Short grace period (low latency systems)
WINDOW TUMBLING (SIZE 5 MINUTES, GRACE PERIOD 1 MINUTE)

-- Medium grace period (typical distributed systems)
WINDOW TUMBLING (SIZE 1 HOUR, GRACE PERIOD 15 MINUTES)

-- Long grace period (high latency or multi-region)
WINDOW TUMBLING (SIZE 1 DAY, GRACE PERIOD 4 HOURS)

Window Retention Tuning

Control how long window state is retained in RocksDB:

-- Set window retention to 12 hours
SET 'ksql.streams.window.store.retention.ms' = '43200000';

-- Now create windowed aggregation
CREATE TABLE hourly_sales WITH (
  KAFKA_TOPIC='hourly_sales',
  VALUE_FORMAT='AVRO'
) AS
SELECT 
  store_id,
  SUM(amount) AS hourly_revenue,
  COUNT(*) AS transaction_count,
  WINDOWSTART AS hour
FROM transactions
WINDOW TUMBLING (SIZE 1 HOUR, GRACE PERIOD 30 MINUTES)
GROUP BY store_id
EMIT CHANGES;

Retention Calculation:

Retention = Window Size + Grace Period + Buffer

Example:
Window: 1 hour (3,600,000 ms)
Grace: 30 minutes (1,800,000 ms)
Buffer: 10 hours (36,000,000 ms)  -- Extra safety margin
Total: 12 hours (43,200,000 ms)

Use Cases:

1. Extended Fraud Detection:

   SET 'ksql.streams.window.store.retention.ms' = '604800000';  -- 7 days
   
   CREATE TABLE suspicious_patterns AS
   SELECT user_id, COUNT(*) AS failed_logins
   FROM login_attempts
   WINDOW TUMBLING (SIZE 1 DAY, GRACE PERIOD 2 HOURS)
   WHERE status = 'FAILED'
   GROUP BY user_id
   HAVING COUNT(*) > 10;
   

2. Multi-Day Trend Analysis:

   SET 'ksql.streams.window.store.retention.ms' = '2592000000';  -- 30 days
   
   CREATE TABLE monthly_user_activity AS
   SELECT user_id, COUNT(*) AS daily_actions
   FROM user_events
   WINDOW TUMBLING (SIZE 1 DAY)
   GROUP BY user_id;
   

3. Compliance Scenarios:

   • Retain window state for audit periods
   • Support regulatory reporting windows
   • Historical data reconstruction

Window Metadata Access

Use WINDOWSTART and WINDOWEND for time-aware processing:

SELECT 
  user_id,
  COUNT(*) AS event_count,
  WINDOWSTART AS window_start_ms,
  WINDOWEND AS window_end_ms,
  FROM_UNIXTIME(WINDOWSTART) AS window_start_readable,
  FROM_UNIXTIME(WINDOWEND) AS window_end_readable,
  (WINDOWEND - WINDOWSTART) / 1000 AS window_size_seconds
FROM user_events
WINDOW TUMBLING (SIZE 1 HOUR)
GROUP BY user_id
EMIT CHANGES;

Result:

+----------+-------------+------------------+----------------+-------------------------+-----------------------+---------------------+
| USER_ID  | EVENT_COUNT | WINDOW_START_MS  | WINDOW_END_MS  | WINDOW_START_READABLE   | WINDOW_END_READABLE   | WINDOW_SIZE_SECONDS |
+----------+-------------+------------------+----------------+-------------------------+-----------------------+---------------------+
| user123  | 45          | 1738407600000    | 1738411200000  | 2025-02-01 10:00:00    | 2025-02-01 11:00:00  | 3600                |
+----------+-------------+------------------+----------------+-------------------------+-----------------------+---------------------+

Advanced Window Patterns

Pattern 1: Nested Aggregations

-- Calculate hourly averages, then daily maximums
CREATE TABLE hourly_temps AS
SELECT sensor_id, AVG(temperature) AS avg_temp, WINDOWSTART AS hour
FROM sensor_readings
WINDOW TUMBLING (SIZE 1 HOUR)
GROUP BY sensor_id;

CREATE TABLE daily_max_temps AS
SELECT sensor_id, MAX(avg_temp) AS max_hourly_avg
FROM hourly_temps
WINDOW TUMBLING (SIZE 1 DAY)
GROUP BY sensor_id;

Pattern 2: Conditional Windowing

-- Count high-value transactions per hour
SELECT 
  merchant_id,
  COUNT(*) AS high_value_count,
  SUM(amount) AS total_amount
FROM transactions
WHERE amount > 1000  -- Filter before windowing
WINDOW TUMBLING (SIZE 1 HOUR)
GROUP BY merchant_id
HAVING COUNT(*) > 10  -- Filter after aggregation
EMIT CHANGES;

Pattern 3: Multi-Level Windows

-- 5-minute windows with 1-hour monitoring
SELECT 
  region,
  COUNT(*) AS request_count,
  WINDOWSTART AS five_min_window
FROM api_requests
WINDOW TUMBLING (SIZE 5 MINUTES, GRACE PERIOD 1 MINUTE)
GROUP BY region
EMIT CHANGES;

Best Practices for Windowing

1. Match Grace Periods to Expected Latency

2. Monitor State Store Memory Usage

-- Check state store sizes
DESCRIBE EXTENDED user_sessions;

-- Look for "Local state store names" and "Storage directory"
-- Monitor RocksDB metrics via JMX

Warning Signs:

• State store size growing unbounded
• Out of memory errors
• Slow aggregation queries
• High disk I/O

Solutions:

• ✅ Reduce window retention
• ✅ Increase JVM heap size
• ✅ Use shorter window sizes
• ✅ Archive old windows to external storage

3. Use WINDOWSTART/WINDOWEND Metadata

Always include window boundaries for:

• Debugging and troubleshooting
• Downstream time-based processing
• Result verification
• Time-series visualization

-- ✅ Good: Include window metadata
SELECT user_id, COUNT(*), WINDOWSTART, WINDOWEND ...

-- ❌ Bad: Missing window context
SELECT user_id, COUNT(*) ...

4. Consider Retention Requirements

Before setting retention:

Ask:

1. How long do consumers need to query historical windows?
2. What's the maximum acceptable late event delay?
3. Are there regulatory/compliance requirements?
4. What's the available disk space?

Example Decision Tree:

Real-time dashboard (last hour only):
  → Retention = 2 hours (window + grace + small buffer)

Fraud detection (7-day patterns):
  → Retention = 10 days (7 days + grace + buffer)

Compliance audit (30-day windows):
  → Retention = 35 days minimum

5. Test with Out-of-Order Data

Always test window behavior with:

• Late-arriving events
• Duplicate events
• Events with future timestamps
• Network partition scenarios

-- Insert test events with varying timestamps
INSERT INTO test_stream (user_id, event_time, event_type) VALUES
  ('u1', 1738407600000, 'click'),   -- On time
  ('u1', 1738407300000, 'click'),   -- 5 min late
  ('u1', 1738411200000, 'click'),   -- Future timestamp
  ('u1', 1738407600000, 'click');   -- Duplicate

Key Takeaways

1. COUNT, SUM, AVG, MIN, MAX: Standard aggregations
2. TUMBLING: Fixed, non-overlapping windows
3. HOPPING: Fixed, overlapping windows
4. SESSION: Activity-based windows
5. GROUP BY: Required for aggregations

What's Next?

Next Module: 10.6.4 – Stateful Processing in ksqlDB