5.3 Windowed Aggregations

Module Overview

Goal: Combine state + windows once — the authoritative source for windowed aggregations.

What You'll Learn:

• Windowed keys and their structure
• Suppression (controlling output)
• Late-arriving data handling
• Emit strategies (early, on-time, late)
• Final vs intermediate results
• Practical windowed aggregation patterns

What's NOT Covered Here:

• Window types (covered in Module 5.2)
• Basic aggregations (covered in Module 5.1)
• State store internals (covered in Module 6.1)

Key Insight: Windowed aggregations combine the power of state with time boundaries to create time-based analytics.

1. Windowed Aggregations Basics

What is a Windowed Aggregation?

Combine grouping by key + grouping by time window + aggregation function:

Regular aggregation:
├─ Group by key
└─ Aggregate (count, sum, etc.)
└─ Result: KTable<K, V>

Windowed aggregation:
├─ Group by key
├─ Group by time window
└─ Aggregate (count, sum, etc.)
└─ Result: KTable<Windowed<K>, V>

Example: Clicks per User per Hour

KStream<String, ClickEvent> clicks = builder.stream("clicks");

KTable<Windowed<String>, Long> clicksPerHour = clicks
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(
        Duration.ofHours(1),
        Duration.ofMinutes(5)))
    .count();

// Result: Count per user per hour
// Key: Windowed<userId> (contains user + window)
// Value: count

2. Windowed Keys

Structure of Windowed<K>

Windowed<String> windowedKey = /* ... */;

// Extract components
String userId = windowedKey.key();           // Original key
Window window = windowedKey.window();        // Window object
long start = window.start();                 // Window start (ms)
long end = window.end();                     // Window end (ms)

Example: Processing Windowed Results

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

windowedCounts.toStream().foreach((windowedKey, count) -> {
    String user = windowedKey.key();
    long windowStart = windowedKey.window().start();
    long windowEnd = windowedKey.window().end();
    
    System.out.printf("User %s had %d clicks from %s to %s%n",
        user,
        count,
        Instant.ofEpochMilli(windowStart),
        Instant.ofEpochMilli(windowEnd)
    );
});

// Output:
// User alice had 15 clicks from 2024-01-15T10:00:00Z to 2024-01-15T11:00:00Z
// User bob had 8 clicks from 2024-01-15T10:00:00Z to 2024-01-15T11:00:00Z

3. Windowed Count

Syntax

KTable<Windowed<K>, Long> count()

What it does: Count records per key per window.

Example: Tumbling Window Count

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

// Count events per user per 5-minute window
KTable<Windowed<String>, Long> counts = events
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(
        Duration.ofMinutes(5),
        Duration.ofMinutes(1)))
    .count();

// Input:
// 10:01 - user1, event
// 10:02 - user1, event
// 10:03 - user2, event
// 10:06 - user1, event  ← New window!

// Output (KTable):
// Windowed<user1>[10:00-10:05] → 2
// Windowed<user2>[10:00-10:05] → 1
// Windowed<user1>[10:05-10:10] → 1

4. Windowed Reduce

Syntax

KTable<Windowed<K>, V> reduce(Reducer<V> reducer)

What it does: Reduce values within each window.

Example: Sum per Window

KStream<String, Integer> sales = builder.stream("sales");

// Sum sales per product per hour
KTable<Windowed<String>, Integer> hourlySales = sales
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1)))
    .reduce((sales1, sales2) -> sales1 + sales2);

// Input (product-A):
// 10:15 → 100
// 10:30 → 50
// 10:45 → 75

// Output:
// Windowed<product-A>[10:00-11:00] → 225

5. Windowed Aggregate

Syntax

KTable<Windowed<K>, VR> aggregate(
    Initializer<VR> initializer,
    Aggregator<K, V, VR> aggregator,
    Materialized<K, VR, WindowStore<Bytes, byte[]>> materialized
)

What it does: Custom aggregation into custom result type per window.

Example: Statistics per Window

KStream<String, Integer> values = builder.stream("values");

KTable<Windowed<String>, Stats> windowedStats = values
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(
        Duration.ofMinutes(10),
        Duration.ofMinutes(2)))
    .aggregate(
        // Initializer
        () -> new Stats(0, 0, 0, 0),
        
        // Aggregator
        (key, value, stats) -> {
            stats.incrementCount();
            stats.addValue(value);
            stats.updateMin(value);
            stats.updateMax(value);
            return stats;
        },
        
        // Materialized
        Materialized.with(Serdes.String(), statsSerde)
    );

// Result: Stats (count, sum, min, max) per key per 10-minute window

6. Suppression: Controlling Output

The Problem

Without suppression:

Window [10:00-10:05]:
├─ Event 1 arrives → emit count=1
├─ Event 2 arrives → emit count=2
├─ Event 3 arrives → emit count=3
└─ Event 4 arrives → emit count=4

Output: 4 updates for same window
└─ Downstream system receives intermediate results
└─ Extra processing, network traffic

What is Suppression?

Suppression delays emitting results until a condition is met (e.g., window closes).

With suppression:

Window [10:00-10:05]:
├─ Event 1 arrives → buffer (don't emit)
├─ Event 2 arrives → buffer (don't emit)
├─ Event 3 arrives → buffer (don't emit)
├─ Event 4 arrives → buffer (don't emit)
└─ Window closes → emit count=4

Output: 1 final result
└─ Downstream sees only complete result

Suppression Syntax

KTable<Windowed<K>, V> suppress(Suppressed<Windowed<K>> suppression)

Example: Suppress Until Window Closes

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

KTable<Windowed<String>, Long> finalCounts = events
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(
        Duration.ofMinutes(5),
        Duration.ofMinutes(1)))
    .count()
    .suppress(Suppressed.untilWindowCloses(
        Suppressed.BufferConfig.unbounded()));

// Only emit when window is finalized (after grace period)

Suppression Strategies

// 1. Suppress until window closes (most common)
Suppressed.untilWindowCloses(BufferConfig.unbounded())

// 2. Suppress until time limit
Suppressed.untilTimeLimit(
    Duration.ofMinutes(1),
    BufferConfig.maxRecords(1000))

// 3. Bounded buffer (prevent memory issues)
Suppressed.untilWindowCloses(
    BufferConfig.maxRecords(10000).emitEarlyWhenFull())

Suppression Trade-offs

WITH SUPPRESSION:
✅ Fewer downstream updates
✅ Only final results
✅ Cleaner output
❌ Higher latency (wait for window to close)
❌ More memory (buffer updates)

WITHOUT SUPPRESSION:
✅ Lower latency (immediate results)
✅ Less memory
❌ Many intermediate updates
❌ Downstream must handle updates

7. Handling Late-Arriving Data

Late Data Scenarios

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

Event timeline:
10:01 - Event A arrives (timestamp 10:02) ✅ On time
10:04 - Event B arrives (timestamp 10:03) ✅ On time
10:06 - Event C arrives (timestamp 10:04) ✅ Within grace (10:05 + 2)
10:08 - Window closes (10:05 + 2 + 1)
10:09 - Event D arrives (timestamp 10:03) ❌ Too late (dropped)

Grace Period Configuration

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

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

Window Retention Configuration

Problem: By default, window state is retained forever (or until default retention), which can cause state to grow unbounded.

Solution: Explicitly configure retention to control how long window state is kept:

// Default retention: window size + grace period
// For 5-min window + 2-min grace = 7 minutes retention

// Explicit retention using .retain()
TimeWindows window = TimeWindows
    .ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(2))
    .retain(Duration.ofHours(2));  // Keep state for 2 hours max

// Use in aggregation
stream
    .groupByKey()
    .windowedBy(window)
    .count();

Retention Rules:

Default Retention = window size + grace period + 24 hours (default)

Custom Retention:
├─ MUST be >= window size + grace period
├─ Controls when state store deletes old windows
└─ Balances state size vs. late data handling

Example: Optimal Retention for Different Scenarios

// Scenario 1: Real-time dashboard (keep recent data only)
TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(1))
    .retain(Duration.ofMinutes(10));  // Only 10 minutes of state

// Scenario 2: Hourly reports (need more history for reprocessing)
TimeWindows.ofSizeAndGrace(Duration.ofHours(1), Duration.ofMinutes(10))
    .retain(Duration.ofHours(6));  // 6 hours of state

// Scenario 3: Daily aggregations (long retention for backfill)
TimeWindows.ofSizeAndGrace(Duration.ofDays(1), Duration.ofHours(1))
    .retain(Duration.ofDays(7));  // 7 days of state

Impact on State Size:

Without explicit retention:
├─ State grows indefinitely
├─ Disk usage increases over time
└─ Eventually causes OOM or disk full

With explicit retention:
├─ Old windows automatically purged
├─ State size bounded
├─ Predictable resource usage

Trade-off:
├─ Shorter retention = Less state, less late data handling
└─ Longer retention = More state, better late data handling

Best Practice: Always set explicit .retain() to avoid unbounded state growth, especially in production. Choose retention based on your late data tolerance and state size constraints.

Late Data Example

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

KTable<Windowed<String>, Long> counts = events
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(
        Duration.ofMinutes(5),
        Duration.ofMinutes(2)))  // Accept late data for 2 minutes
    .count();

// Event flow:
// 10:01 - event (ts=10:02) → counted in [10:00-10:05]
// 10:06 - event (ts=10:04) → counted in [10:00-10:05] (within grace)
// 10:08 - event (ts=10:03) → DROPPED (beyond grace period)

Detecting Dropped Records

// Log late records
events
    .process(() -> new Processor<String, Event>() {
        @Override
        public void process(String key, Event value) {
            long eventTime = value.getTimestamp();
            long processingTime = context().timestamp();
            long lateness = processingTime - eventTime;
            
            if (lateness > Duration.ofMinutes(2).toMillis()) {
                log.warn("Late event: key={}, lateness={}ms", key, lateness);
            }
            
            forward(key, value);
        }
    });

8. Emit Strategies

When to Emit Results?

THREE TIMING OPTIONS:

1. Early (eager):
   └─ Emit as soon as data arrives
   └─ Low latency, many updates

2. On-time (default):
   └─ Emit when window closes (grace period expires)
   └─ Balanced

3. Late:
   └─ Only emit after final window closure
   └─ Highest latency, fewest updates

Configuring Emit Strategy (via Suppression)

// Default: emit every update (early)
windowedStream.count();

// On-time: emit only when window closes
windowedStream
    .count()
    .suppress(Suppressed.untilWindowCloses(BufferConfig.unbounded()));

// Custom: emit periodically
windowedStream
    .count()
    .suppress(Suppressed.untilTimeLimit(
        Duration.ofSeconds(30),
        BufferConfig.maxRecords(1000)));

9. Session Window Aggregations

Unique Challenges

Session windows have dynamic boundaries:

Sessions can MERGE:

Initial state:
Session A: 10:00 - 10:03 (3 events)
Session B: 10:10 - 10:12 (2 events)

Late event at 10:06 arrives:
├─ Fills gap between sessions
└─ Sessions merge!

Merged session: 10:00 - 10:12 (6 events)

Session Window Count

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

KTable<Windowed<String>, Long> sessionCounts = events
    .groupByKey()
    .windowedBy(SessionWindows.ofInactivityGapAndGrace(
        Duration.ofMinutes(5),    // Inactivity gap
        Duration.ofMinutes(1)))   // Grace period
    .count();

// Sessions close after 5 minutes of inactivity
// Late events within 1 minute can extend/merge sessions

Session Aggregation with Merging

KTable<Windowed<String>, SessionStats> sessionStats = events
    .groupByKey()
    .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(
        Duration.ofMinutes(5)))
    .aggregate(
        // Initializer
        SessionStats::new,
        
        // Aggregator: add event to session
        (key, event, stats) -> {
            stats.addEvent(event);
            return stats;
        },
        
        // Merger: combine two sessions
        (key, stats1, stats2) -> {
            return stats1.merge(stats2);
        },
        
        Materialized.with(Serdes.String(), sessionStatsSerde)
    );

Note: Session aggregations require a merger function!

10. Writing Windowed Results to Topics

Converting to Stream

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

// Convert to stream for writing
KStream<Windowed<String>, Long> stream = windowedCounts.toStream();

// Problem: Windowed key can't be written directly to Kafka
// Solution: flatten to string key

Flattening Windowed Keys

windowedCounts
    .toStream()
    .map((windowedKey, count) -> {
        String key = windowedKey.key();
        long start = windowedKey.window().start();
        long end = windowedKey.window().end();
        
        // Create composite key
        String flatKey = String.format("%s|%d|%d", key, start, end);
        
        return KeyValue.pair(flatKey, count);
    })
    .to("windowed-counts-output");

// Output:
// Key: "user1|1704538800000|1704539100000"
// Value: 15

Alternative: Custom Serde

// Define custom serde for Windowed<String>
Serde<Windowed<String>> windowedSerde = 
    new WindowedSerdes.TimeWindowedSerde<>(Serdes.String());

windowedCounts
    .toStream()
    .to("windowed-counts",
        Produced.with(windowedSerde, Serdes.Long()));

11. Common Patterns

Pattern 1: Hourly Metrics

// Count events per user per hour
events
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(
        Duration.ofHours(1),
        Duration.ofMinutes(5)))
    .count()
    .suppress(Suppressed.untilWindowCloses(BufferConfig.unbounded()))
    .toStream()
    .to("hourly-metrics");

Pattern 2: Moving Average

// 10-minute average, updated every 2 minutes
values
    .groupByKey()
    .windowedBy(TimeWindows
        .ofSizeAndGrace(Duration.ofMinutes(10), Duration.ofMinutes(1))
        .advanceBy(Duration.ofMinutes(2)))
    .aggregate(
        () -> new Average(0, 0.0),
        (key, value, avg) -> avg.add(value),
        Materialized.with(Serdes.String(), averageSerde)
    );

Pattern 3: Session Analytics

// User session statistics
clickstreams
    .groupByKey()
    .windowedBy(SessionWindows.ofInactivityGapAndGrace(
        Duration.ofMinutes(30),
        Duration.ofMinutes(5)))
    .aggregate(
        SessionMetrics::new,
        (userId, click, metrics) -> metrics.addClick(click),
        (userId, metrics1, metrics2) -> metrics1.merge(metrics2),
        Materialized.with(Serdes.String(), sessionMetricsSerde)
    )
    .suppress(Suppressed.untilWindowCloses(BufferConfig.unbounded()));

Pattern 4: Late Data Monitoring

// Track and log late arrivals
KStream<String, Event> events = builder.stream("events");

events.foreach((key, event) -> {
    long eventTime = event.getTimestamp();
    long processingTime = System.currentTimeMillis();
    long lateness = processingTime - eventTime;
    
    if (lateness > Duration.ofMinutes(1).toMillis()) {
        metrics.recordLateArrival(key, lateness);
    }
});

// Then do windowed aggregation
KTable<Windowed<String>, Long> counts = events
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(
        Duration.ofMinutes(5),
        Duration.ofMinutes(1)))
    .count();

12. Performance Considerations

State Size

Windowed aggregations store:
├─ One entry per key per window
└─ State grows with: (# unique keys) × (# active windows)

Example:
├─ 100,000 unique users
├─ 5-minute windows, 1 hour retention
├─ 12 windows active (60 min / 5 min)
└─ State: 100,000 × 12 = 1.2M entries

Mitigation:
✅ Use appropriate window sizes
✅ Set retention correctly
✅ Use suppression to reduce updates

Grace Period Impact

LONGER GRACE PERIOD:
❌ Windows stay open longer
❌ More state
❌ Higher latency
✅ More accurate (accept more late data)

SHORTER GRACE PERIOD:
✅ Windows close faster
✅ Less state
✅ Lower latency
❌ Drop more late data

Suppression Memory

Suppression buffers updates in memory:

Without suppression:
├─ Update immediately written to changelog
├─ Low memory, high I/O

With suppression:
├─ Updates buffered until emission
├─ Higher memory, lower I/O

Use bounded buffers in production:
Suppressed.untilWindowCloses(
    BufferConfig.maxRecords(10000).emitEarlyWhenFull())

13. Testing Windowed Aggregations

Using TopologyTestDriver

@Test
public void testWindowedCount() {
    TopologyTestDriver driver = new TopologyTestDriver(topology, props);
    
    TestInputTopic<String, Event> input = 
        driver.createInputTopic("events", /* serdes */);
    
    TestOutputTopic<String, Long> output = 
        driver.createOutputTopic("windowed-counts", /* serdes */);
    
    // Send events in same window
    input.pipeInput("user1", new Event(), 1000L);  // 00:00:01
    input.pipeInput("user1", new Event(), 2000L);  // 00:00:02
    input.pipeInput("user1", new Event(), 3000L);  // 00:00:03
    
    // Advance time beyond window + grace period
    input.pipeInput("user1", new Event(), 400000L); // 00:06:40
    
    // Read result
    KeyValue<String, Long> result = output.readKeyValue();
    assertEquals(3L, result.value);  // 3 events in window
}

Testing covered in Module 11.1.

14. Mental Model Summary

Windowed Aggregations

COMPONENTS:

1. Grouping
   └─ groupByKey() or groupBy()

2. Windowing
   └─ windowedBy(TimeWindows/SessionWindows)

3. Aggregation
   └─ count() / reduce() / aggregate()

4. Suppression (optional)
   └─ suppress() to control output

RESULT:
└─ KTable<Windowed<K>, V>
   └─ Key contains: original key + window
   └─ Value: aggregated result

Key Decisions

1. Window Type
   ├─ Tumbling: fixed, non-overlapping
   ├─ Hopping: fixed, overlapping
   └─ Session: dynamic, activity-based

2. Grace Period
   ├─ None: strict, drop all late data
   ├─ Short: some tolerance
   └─ Long: high tolerance, higher latency

3. Suppression
   ├─ None: emit every update
   ├─ Until window closes: emit only final
   └─ Custom: balanced approach

4. Output Format
   ├─ Keep windowed key: for further processing
   └─ Flatten key: for external systems

15. Questions You Should Be Able to Answer

After this module, you should be able to answer:

✅ What is a windowed aggregation?
✅ What does Windowed<K> contain?
✅ When should I use suppression?
✅ How do grace periods affect results?
✅ How do I handle late-arriving data?
✅ What's special about session window aggregations?
✅ How do I write windowed results to a topic?

16. What's Next?

Now that you understand windowed aggregations, the next modules will cover:

• Module 6.1: State stores — how windowed state is physically stored
• Module 6.2: Changelogs — durability and backup
• Module 6.3: Restore & recovery — handling failures

Summary

You learned:

• ✅ Windowed aggregations combine grouping, windowing, and aggregation
• ✅ Windowed<K> contains original key + window start/end times
• ✅ Suppression controls when results are emitted (reduces downstream updates)
• ✅ Grace periods allow late data with latency trade-off
• ✅ Session windows require special merger function
• ✅ Writing results requires flattening windowed keys
• ✅ Performance depends on window size, grace period, and suppression strategy

Next up: Module 6.1 — State Stores Explained — Learn how Kafka Streams physically stores and manages state.