Windowed operations in Kafka - Mini Project: Build & Apply

Windowed Operations with Kafka Streams

📖 Scenario: You are working with a stream of sales data coming from a Kafka topic. Each record contains a product name and the number of units sold. You want to calculate the total units sold for each product in 1-minute time windows.

🎯 Goal: Build a Kafka Streams application that reads sales data, groups it by product name, applies a 1-minute tumbling window, sums the units sold per product in each window, and outputs the results.

📋 What You'll Learn

Create a Kafka Streams builder and define the input stream from topic sales.

Create a 1-minute tumbling windowed aggregation grouped by product name.

Sum the units sold per product in each window.

Print the windowed aggregation results to the console.

💡 Why This Matters

🌍 Real World

Windowed operations are used in real-time analytics to summarize data over fixed time periods, such as counting sales per minute or monitoring sensor data.

💼 Career

Understanding windowed aggregations in Kafka Streams is essential for building scalable, real-time data processing applications in roles like data engineer or streaming developer.

1

Create the initial stream from the sales topic

Create a Kafka Streams builder called builder and define a stream called salesStream from the topic "sales" with key type String and value type Integer.

Kafka

// Create a StreamsBuilder and define salesStream from topic "sales"
// Your code here

Hint

Use StreamsBuilder to create the builder and builder.stream("sales") to get the stream.

2

Define a 1-minute tumbling window for grouping

Create a variable called windowSize of type Duration and set it to 1 minute. Then create a TimeWindows object called tumblingWindow using TimeWindows.ofSizeWithNoGrace(windowSize).

Kafka

StreamsBuilder builder = new StreamsBuilder();
KStream salesStream = builder.stream("sales");
// Define a 1-minute Duration called windowSize
// Define a tumbling TimeWindows called tumblingWindow
// Your code here

Hint

Use Duration.ofMinutes(1) for the window size and TimeWindows.ofSizeWithNoGrace(windowSize) for the tumbling window.

3

Group by product and sum units sold in each window

Group salesStream by key using groupByKey(), then apply a windowed aggregation using windowedBy(tumblingWindow) and sum the values using reduce(Integer::sum). Store the result in a variable called windowedSales of type KTable<Windowed<String>, Integer>.

Kafka

StreamsBuilder builder = new StreamsBuilder();
KStream salesStream = builder.stream("sales");
Duration windowSize = Duration.ofMinutes(1);
TimeWindows tumblingWindow = TimeWindows.ofSizeWithNoGrace(windowSize);
// Group salesStream by key, window by tumblingWindow, and sum values
// Your code here

Hint

Use groupByKey() then windowedBy(tumblingWindow) and reduce(Integer::sum) to sum units sold per window.

4

Print the windowed aggregation results

Use windowedSales.toStream() and then foreach() to print each window's product name, window start time, and total units sold. Use System.out.println inside the lambda to print in the format: "Product: {key}, Window start: {windowStart}, Total units: {value}".

Kafka

StreamsBuilder builder = new StreamsBuilder();
KStream salesStream = builder.stream("sales");
Duration windowSize = Duration.ofMinutes(1);
TimeWindows tumblingWindow = TimeWindows.ofSizeWithNoGrace(windowSize);
KTable, Integer> windowedSales = salesStream
    .groupByKey()
    .windowedBy(tumblingWindow)
    .reduce(Integer::sum);
// Print each windowed aggregation result
// Your code here

Hint

Use toStream() and foreach() to print each windowed result with System.out.println.

Practice

1. What is the main purpose of windowed operations in Kafka Streams?

easy

A. To compress data for storage

B. To group data into time intervals for easier analysis

C. To encrypt data streams

D. To sort data alphabetically

Solution

1. Step 1: Understand windowed operations

   Windowed operations group data based on time intervals, making it easier to analyze events occurring within those intervals.

2. Step 2: Compare options

   Sorting, encrypting, and compressing are unrelated to windowing, which focuses on time-based grouping.

3. Final Answer:

   To group data into time intervals for easier analysis -> Option B

4. Quick Check:

   Windowed operations = time grouping [OK]

Hint: Windowed means grouping by time intervals [OK]

Common Mistakes:

• Confusing windowing with sorting
• Thinking windowing encrypts data
• Assuming windowing compresses data

2. Which of the following is the correct syntax to create a tumbling window of 5 minutes in Kafka Streams DSL?

easy

A. SlidingWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5))

B. SessionWindows.of(Duration.ofMinutes(5))

C. TimeWindows.of(Duration.ofMinutes(5))

D. Windowed.of(Duration.ofMinutes(5))

Solution

1. Step 1: Identify tumbling window syntax

   Tumbling windows in Kafka Streams are created using TimeWindows.of() with a duration.

2. Step 2: Check options

   SessionWindows is for session windows, SlidingWindows is for sliding windows, and Windowed is not a window type class.

3. Final Answer:

   TimeWindows.of(Duration.ofMinutes(5)) -> Option C

4. Quick Check:

   Tumbling window = TimeWindows.of() [OK]

Hint: Tumbling windows use TimeWindows.of() syntax [OK]

Common Mistakes:

• Using SessionWindows for tumbling windows
• Confusing SlidingWindows with tumbling windows
• Using non-existent Windowed class

3. Given the following Kafka Streams code snippet, what will be the output count for the window ending at timestamp 60000?

stream.groupByKey()
      .windowedBy(TimeWindows.of(Duration.ofSeconds(30)))
      .count()
      .toStream()
      .foreach((windowedKey, count) -> System.out.println(windowedKey.window().end() + ": " + count));

Assume events with keys arrive at timestamps 10000, 20000, 35000, 40000, and 55000.

medium

A. 3

B. 4

C. 2

D. 5

Solution

1. Step 1: Identify window intervals

   Windows are 30 seconds long: [0-30000), [30000-60000), etc. Events at 10000 and 20000 fall in first window; 35000, 40000, 55000 fall in second window ending at 60000.

2. Step 2: Count events in window ending at 60000

   Events at 35000, 40000, and 55000 are inside the window ending at 60000, so count is 3.

3. Final Answer:

   3 -> Option A

4. Quick Check:

   Events in 30s window = 3 [OK]

Hint: Count events in the correct time window interval [OK]

Common Mistakes:

• Counting events from previous window
• Misunderstanding window boundaries
• Including events outside the window

4. Identify the error in this Kafka Streams windowed count code snippet:

stream.groupByKey()
      .windowedBy(SessionWindows.of(Duration.ofMinutes(1)))
      .count()
      .toStream()
      .foreach((windowedKey, count) -> System.out.println(windowedKey.window().start() + ": " + count));

medium

A. windowedBy() cannot be used with groupByKey()

B. toStream() is not available after count()

C. count() must be called before windowedBy()

D. SessionWindows.of() requires a gap duration, not a window size

Solution

1. Step 1: Check SessionWindows.of() usage

   SessionWindows require a gap duration (inactivity gap), not a fixed window size in of().

2. Step 2: Validate other method calls

   groupByKey() works with windowedBy(), count() is correctly placed, and toStream() is valid after count().

3. Final Answer:

   SessionWindows.of() requires a gap duration, not a window size -> Option D

4. Quick Check:

   SessionWindows need gap duration [OK]

Hint: SessionWindows need gap duration, not fixed size [OK]

Common Mistakes:

• Using fixed duration in SessionWindows.of()
• Misplacing count() before windowedBy()
• Assuming toStream() is invalid after count()

5. You want to count user clicks in overlapping 1-minute windows that advance every 30 seconds using Kafka Streams. Which window type and configuration should you use?

hard

A. Use TimeWindows.of(Duration.ofMinutes(1)).advanceBy(Duration.ofSeconds(30))

B. Use SlidingWindows.ofTimeDifferenceAndGrace(Duration.ofMinutes(1), Duration.ofSeconds(0)).advanceBy(Duration.ofSeconds(30))

C. Use SessionWindows.withGap(Duration.ofSeconds(30))

D. Use TimeWindows.of(Duration.ofSeconds(30)).advanceBy(Duration.ofMinutes(1))

Solution

1. Step 1: Identify overlapping window type

   Overlapping windows that slide are created with TimeWindows.of() and advanceBy() to set the slide interval.

2. Step 2: Match configuration to requirement

   Window size is 1 minute, sliding every 30 seconds, so TimeWindows.of(1 minute).advanceBy(30 seconds) fits perfectly.

3. Final Answer:

   Use TimeWindows.of(Duration.ofMinutes(1)).advanceBy(Duration.ofSeconds(30)) -> Option A

4. Quick Check:

   Sliding windows = TimeWindows with advanceBy [OK]

Hint: Sliding windows use TimeWindows with advanceBy() [OK]

Common Mistakes:

• Using SlidingWindows class which is deprecated
• Confusing session windows with sliding windows
• Swapping window size and advanceBy durations