5.2 Windowing Concepts & Types

Module Overview

Goal: Centralize all window theory in one place.

What You'll Learn:

• Why windows exist in stream processing
• Tumbling windows
• Hopping windows
• Sliding windows
• Session windows
• Grace periods and late data
• Window retention

What's NOT Covered Here:

• Aggregation code (covered in Module 5.3)
• State store internals (covered in Module 6.1)
• Retention configuration details (covered in Module 6.1)

Key Insight: Windows provide a way to bound state over time, making unbounded streams manageable.

1. Why Windows Exist

The Unbounded Stream Problem

Problem: How to aggregate an infinite stream?

Stream of events: [e1, e2, e3, e4, ... e∞]

Question: "What's the total?"
Answer: Infinite accumulation → unbounded state → runs out of memory

Solution: Window the data
└─ Break infinite stream into finite chunks
└─ Aggregate within each chunk
└─ State becomes bounded

Real-World Examples

Instead of asking:
❌ "Total clicks ever" (unbounded)

Ask:
✅ "Total clicks in last hour" (bounded by time)
✅ "Total clicks per day" (bounded by time)
✅ "Total clicks per user session" (bounded by inactivity)

Windows make these questions answerable

2. Event Time vs Processing Time

Two Notions of Time

EVENT TIME:
└─ When event actually happened
└─ Embedded in the event itself
└─ Example: order.timestamp = "2024-01-15 10:30:00"

PROCESSING TIME:
└─ When event is processed by Streams app
└─ System time when record consumed
└─ Example: System.currentTimeMillis()

Which to Use?

Event Time (preferred):
✅ Accurate business logic
✅ Replay gives same results
✅ Handles out-of-order events
⚠️ Requires timestamps in data

Processing Time:
✅ Simple (no timestamp needed)
✅ Lower latency
❌ Replay gives different results
❌ Out-of-order events misplaced

Kafka Streams uses EVENT TIME by default

Extracting Event Time

// Tell Kafka Streams how to extract timestamp
stream
    .selectKey(...)
    .groupByKey()
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
    .count();

// Default: uses record timestamp from Kafka
// Can override with custom TimestampExtractor

3. Tumbling Windows

Definition

Tumbling windows are fixed-size, non-overlapping windows.

Time →
┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐
│Window 1│ │Window 2│ │Window 3│ │Window 4│
│ 0-5min │ │ 5-10min│ │10-15min│ │15-20min│
└────────┘ └────────┘ └────────┘ └────────┘

Each event belongs to exactly ONE window
No overlap between windows

Characteristics

TUMBLING WINDOWS:

1. Fixed Size
   └─ All windows same duration

2. Non-Overlapping
   └─ Each event in exactly one window

3. Aligned to Epoch
   └─ Windows start at fixed intervals from epoch

4. Use Case
   └─ Periodic reports, dashboards

Example: 5-Minute Tumbling Windows

Window 1: 10:00 - 10:05
Window 2: 10:05 - 10:10
Window 3: 10:10 - 10:15

Event at 10:03 → Window 1
Event at 10:07 → Window 2
Event at 10:14 → Window 3

Each event goes to exactly one window

Code Example

KStream<String, Long> pageViews = builder.stream("page-views");

KTable<Windowed<String>, Long> windowedCounts = pageViews
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
    .count();

// Output: count per user per 5-minute window
// Key: Windowed<userId>
// Value: count

4. Hopping Windows

Definition

Hopping windows are fixed-size, overlapping windows with a defined advance interval.

Time →
┌────────┐
│Window 1│   Window size: 10 min
│ 0-10   │   Advance:    5 min
└────────┘
     ┌────────┐
     │Window 2│
     │ 5-15   │
     └────────┘
          ┌────────┐
          │Window 3│
          │10-20   │
          └────────┘

Windows overlap
Each event can be in multiple windows

Characteristics

HOPPING WINDOWS:

1. Fixed Size
   └─ Window duration constant

2. Fixed Advance
   └─ New window starts every advance interval

3. Overlapping
   └─ Events can be in multiple windows

4. Use Case
   └─ Moving averages, sliding statistics

Example: 10-Minute Windows, 5-Minute Hop

Window 1: 10:00 - 10:10
Window 2: 10:05 - 10:15
Window 3: 10:10 - 10:20

Event at 10:07:
├─ In Window 1 (10:00 - 10:10) ✅
├─ In Window 2 (10:05 - 10:15) ✅
└─ Not in Window 3

Event counted in MULTIPLE windows

Code Example

KStream<String, Long> pageViews = builder.stream("page-views");

KTable<Windowed<String>, Long> windowedCounts = pageViews
    .groupByKey()
    .windowedBy(TimeWindows
        .ofSizeWithNoGrace(Duration.ofMinutes(10))
        .advanceBy(Duration.ofMinutes(5)))
    .count();

// Output: count per user per window
// Windows overlap by 5 minutes

Hopping vs Tumbling

Tumbling: advanceBy = size
├─ No overlap
└─ Each event in one window

Hopping: advanceBy < size
├─ Overlap
└─ Each event in multiple windows

Special case: advanceBy = size → becomes tumbling

5. Sliding Windows

Definition

Sliding windows are continuous windows that slide with each event.

Time →
Events:    E1      E2        E3
           ↓       ↓         ↓
Window:  [---- 5 min ----]
                 [---- 5 min ----]
                          [---- 5 min ----]

Window moves with each new event
No fixed boundaries

Characteristics

SLIDING WINDOWS:

1. Time Difference-Based
   └─ Window = all events within time diff

2. No Fixed Boundaries
   └─ Windows slide continuously

3. Event-Driven
   └─ New window for each event

4. Use Case
   └─ Correlating events (joins, pattern detection)

Example: 5-Minute Sliding Window

Events:
10:00 - Event A
10:02 - Event B
10:04 - Event C
10:07 - Event D

At 10:04 (Event C):
Window includes events from 9:59 - 10:04
└─ Event A (10:00) ✅
└─ Event B (10:02) ✅
└─ Event C (10:04) ✅

At 10:07 (Event D):
Window includes events from 10:02 - 10:07
└─ Event A (10:00) ❌ (too old)
└─ Event B (10:02) ✅
└─ Event C (10:04) ✅
└─ Event D (10:07) ✅

Code Example

// Used primarily for joins
KStream<String, PageView> views = builder.stream("page-views");
KStream<String, AdClick> clicks = builder.stream("ad-clicks");

// Join views with clicks within 5-minute window
KStream<String, ViewWithClick> joined = views.join(
    clicks,
    (view, click) -> new ViewWithClick(view, click),
    JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5))
);

// Covered in Module 8.1

6. Session Windows

Definition

Session windows are dynamic windows based on activity and inactivity gaps.

Time →
Events:  E1  E2    E3              E4  E5
         ↓   ↓     ↓               ↓   ↓
         [---Session 1---]  gap   [Session 2]
                          (5 min)

Window closes after inactivity gap
New activity starts new session

Characteristics

SESSION WINDOWS:

1. Dynamic Duration
   └─ Window size varies based on activity

2. Inactivity Gap
   └─ Session ends after gap with no events

3. Per-Key Sessions
   └─ Each key has own sessions

4. Use Case
   └─ User sessions, clickstreams, batch jobs

Example: 5-Minute Inactivity Gap

User Activity:
10:00 - Click 1
10:02 - Click 2
10:03 - Click 3
10:15 - Click 4  ← 12 min gap!

Sessions:
Session 1: 10:00 - 10:03
├─ Click 1, Click 2, Click 3
└─ Ended: no activity for 5+ minutes

Session 2: 10:15 - ...
├─ Click 4
└─ Still active (or ends after 5 min inactivity)

Code Example

KStream<String, Event> events = builder.stream("user-events");

KTable<Windowed<String>, Long> sessionCounts = events
    .groupByKey()
    .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(
        Duration.ofMinutes(5)))
    .count();

// Output: count of events per session
// Session closes after 5 minutes of inactivity

Session Merging

Important: Sessions can MERGE

Initial:
Session A: 10:00 - 10:03
Session B: 10:10 - 10:15

Late event at 10:05 arrives:
├─ Falls within gap of both sessions
└─ Sessions merge!

Merged Session: 10:00 - 10:15
└─ All events from A + B + late event

7. Grace Periods and Late Data

The Late Data Problem

Event Time vs Processing Time mismatch:

Event A: timestamp=10:00, arrives at 10:01 ✅ On time
Event B: timestamp=10:00, arrives at 10:20 ⚠️ Late!

Question: Should Event B be included in 10:00's window?

Grace Period

Grace period is extra time to wait for late events before finalizing window.

Window: 10:00 - 10:05
Grace period: 2 minutes

Window stays open until: 10:07

Events arriving before 10:07:
├─ Timestamp between 10:00 - 10:05 ✅ Included

Events arriving after 10:07:
├─ Timestamp between 10:00 - 10:05 ❌ Too late (dropped)

Configuring Grace Period

// With grace period (2 minutes)
TimeWindows.ofSizeAndGrace(
    Duration.ofMinutes(5),
    Duration.ofMinutes(2)
);

// No grace period (strict)
TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5));

Grace Period Trade-offs

LONGER GRACE PERIOD:
✅ Accept more late data
✅ More accurate results
❌ Higher latency (wait longer)
❌ More state (keep windows open)

SHORTER GRACE PERIOD:
✅ Lower latency (results sooner)
✅ Less state
❌ Drop late data
❌ Less accurate results

NO GRACE PERIOD:
✅ Lowest latency
❌ No late data handling

8. Window Comparison

JoinWindows vs TimeWindows (Important Distinction)

Critical: For stream-to-stream joins, you MUST use JoinWindows, NOT TimeWindows!

// ❌ WRONG: Can't use TimeWindows for stream joins
ordersStream.join(
    paymentsStream,
    joiner,
    TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5))  // ❌ Won't compile!
);

// ✅ CORRECT: Use JoinWindows for stream joins
ordersStream.join(
    paymentsStream,
    joiner,
    JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5))  // ✅ Correct
);

Why the difference?

TimeWindows (for aggregations):
├─ Groups events by time buckets
├─ Used with: count(), reduce(), aggregate()
└─ Example: "Count events every 5 minutes"

JoinWindows (for stream joins):
├─ Defines time difference tolerance for matching events
├─ Used with: join(), leftJoin(), outerJoin()
└─ Example: "Join events within 5 minutes of each other"

Key Differences:

Example - TimeWindows (Aggregation):

KStream<String, Order> orders = builder.stream("orders");

// Count orders every 5 minutes
orders
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
    .count();

Example - JoinWindows (Stream Join):

KStream<String, Order> orders = builder.stream("orders");
KStream<String, Payment> payments = builder.stream("payments");

// Join orders with payments within 5-minute window
orders.join(
    payments,
    (order, payment) -> new OrderWithPayment(order, payment),
    JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5))
);

Remember: Aggregations use TimeWindows, Joins use JoinWindows. Mixing them up is a common mistake!

9. Window Retention

Why Retention Matters

Problem: Old window state accumulates

Without retention:
├─ Windows from last month still in state
├─ State grows forever
└─ Eventually runs out of disk

With retention:
├─ Old windows automatically deleted
├─ State remains bounded
└─ Disk usage stable

Default Retention

Default retention = window size + grace period

Example:
Window size:    5 minutes
Grace period:   2 minutes
Retention:      7 minutes

After 7 minutes, window state deleted

Custom Retention

TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(2))
    .until(Duration.ofMinutes(10).toMillis());  // Custom retention

Covered in detail in Module 6.1.

10. Choosing the Right Window Type

Decision Tree

Do you need fixed boundaries?
├─ YES → Tumbling or Hopping
│        │
│        └─ Do you need overlap?
│           ├─ NO → Tumbling
│           └─ YES → Hopping
│
└─ NO → Dynamic windows
         │
         └─ Based on inactivity?
            ├─ YES → Session
            └─ NO → Sliding (for joins)

Use Case Examples

TUMBLING WINDOWS:
✅ Hourly sales reports
✅ Daily user counts
✅ Periodic aggregations

HOPPING WINDOWS:
✅ 10-minute average over last 30 minutes
✅ Moving averages
✅ Overlapping statistics

SLIDING WINDOWS:
✅ Stream-stream joins
✅ Correlation within time window
✅ Event pattern detection

SESSION WINDOWS:
✅ User browsing sessions
✅ Shopping cart sessions
✅ Batch job analysis

11. Window Keys

Windowed Key Structure

Aggregations produce: KTable<Windowed<K>, V>

Windowed<K> contains:
├─ Original key (e.g., userId)
├─ Window start time
└─ Window end time

Example:
Key: Windowed<"user123">
├─ user: "user123"
├─ start: 1704538800000 (10:00)
└─ end:   1704539100000 (10:05)

Extracting Window Information

KTable<Windowed<String>, Long> windowedCounts = /* ... */;

windowedCounts.toStream().foreach((windowedKey, count) -> {
    String user = windowedKey.key();
    long start = windowedKey.window().start();
    long end = windowedKey.window().end();
    
    System.out.printf("User %s: %d events from %s to %s%n",
        user, count, formatTime(start), formatTime(end));
});

12. Common Window Patterns

Pattern 1: Fixed-Interval Reporting

// Count events every 5 minutes
stream
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
    .count();

Pattern 2: Moving Average

// 10-minute window, updated every 2 minutes
stream
    .groupByKey()
    .windowedBy(TimeWindows
        .ofSizeWithNoGrace(Duration.ofMinutes(10))
        .advanceBy(Duration.ofMinutes(2)))
    .aggregate(...);

Pattern 3: User Sessions

// Session per user, 30-minute inactivity gap
stream
    .groupByKey()
    .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(
        Duration.ofMinutes(30)))
    .count();

13. Mental Model Summary

Window Types

TUMBLING: [--][--][--][--]
├─ Fixed size, no overlap
└─ Each event in one window

HOPPING:  [----]
            [----]
              [----]
├─ Fixed size, fixed overlap
└─ Each event in multiple windows

SLIDING:  Continuous, event-driven
├─ No fixed boundaries
└─ For joins/correlation

SESSION:  [---] gap [----] gap [--]
├─ Dynamic size
└─ Based on activity/inactivity

Time Concepts

EVENT TIME:
└─ When event happened (in data)
└─ Kafka Streams default

PROCESSING TIME:
└─ When event processed (system time)
└─ Less common, less accurate

GRACE PERIOD:
└─ Extra time to wait for late events
└─ Trade latency for accuracy

14. Questions You Should Be Able to Answer

After this module, you should be able to answer:

✅ Why do we need windows in stream processing?
✅ What's the difference between tumbling and hopping windows?
✅ When would I use session windows?
✅ What is a grace period?
✅ What's the difference between event time and processing time?
✅ How do I choose the right window type?
✅ What happens to late-arriving data?

15. What's Next?

Now that you understand window concepts, the next module will show you how to use them:

• Module 5.3: Windowed aggregations — combining windows with aggregations
• Module 6.1: State stores — how windowed state is physically stored
• Module 8.1: Joins — windowed joins between streams

Summary

You learned:

• ✅ Windows bound unbounded streams to make state manageable
• ✅ Tumbling windows: fixed, non-overlapping (periodic reports)
• ✅ Hopping windows: fixed, overlapping (moving averages)
• ✅ Sliding windows: continuous, event-driven (joins)
• ✅ Session windows: dynamic, inactivity-based (user sessions)
• ✅ Event time (accurate) vs processing time (simple)
• ✅ Grace periods allow late data with latency trade-off
• ✅ Window retention prevents unbounded state growth

Next up: Module 5.3 — Windowed Aggregations — Learn how to combine windows with aggregations for time-based analytics.