Windowed operations in Kafka - Commands & Configuration

Introduction

Windowed operations let you group streaming data into time chunks to analyze events that happen close together. This helps find patterns or counts over short periods instead of the whole stream.

When you want to count how many times a user clicks a button every minute.

When you need to detect spikes in sensor data over 5-minute intervals.

When you want to aggregate sales totals by hour from a continuous stream.

When you want to join two streams but only consider events that happened within the same time window.

When you want to calculate moving averages or trends over recent data chunks.

Config File - windowed_streams.properties

windowed_streams.properties

bootstrap.servers=localhost:9092
application.id=windowed-operations-app
processing.guarantee=exactly_once
cache.max.bytes.buffering=10485760
commit.interval.ms=1000

This configuration file sets up the Kafka Streams application to connect to the Kafka server at localhost:9092. The application.id uniquely identifies the stream processing app. processing.guarantee=exactly_once ensures data is processed without duplication. cache.max.bytes.buffering controls memory buffering for performance. commit.interval.ms sets how often processed results are saved.

Commands

Create a Kafka topic named 'clicks' with 3 partitions to hold streaming click events.

Terminal

kafka-topics --create --topic clicks --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

Expected Output

Created topic clicks.

→

--partitions - Number of partitions for parallelism

→

--replication-factor - Number of copies for fault tolerance

Start a producer to send click events to the 'clicks' topic. You can type messages here to simulate clicks.

Terminal

kafka-console-producer --topic clicks --bootstrap-server localhost:9092

Expected Output

No output (command runs silently)

→

--topic - Topic to send messages to

→

--bootstrap-server - Kafka server address

Run the Kafka Streams app that reads from 'clicks', counts events in 1-minute windows, and writes results to 'clicks-counts'.

Terminal

kafka-streams-application --config windowed_streams.properties --input-topic clicks --output-topic clicks-counts --window-size 60000

Expected Output

Starting Kafka Streams application with window size 60000 ms

Processing events...

→

--config - Configuration file for the streams app

→

--input-topic - Input topic to read events from

→

--output-topic - Output topic to write windowed counts

→

--window-size - Size of the time window in milliseconds

Consume and display the windowed counts from the 'clicks-counts' topic to verify the windowed operation results.

Terminal

kafka-console-consumer --topic clicks-counts --bootstrap-server localhost:9092 --from-beginning

Expected Output

{"window_start":1680000000000,"count":5}

{"window_start":1680000060000,"count":3}

→

--from-beginning - Read all messages from the start

Key Concept

If you remember nothing else from this pattern, remember: windowed operations group streaming data into fixed time chunks to analyze recent events together.

Common Mistakes

Not setting the window size correctly or forgetting to specify it.

Without a window size, Kafka Streams cannot group events by time, so windowed operations won't work.

Always specify the window size in milliseconds when defining windowed operations.

Using a topic with too few partitions for the expected load.

Too few partitions limit parallel processing and can cause bottlenecks.

Create topics with enough partitions to handle your data volume and processing needs.

Not consuming from the output topic to verify results.

Without checking output, you won't know if windowed operations are working as expected.

Use a console consumer to read from the output topic and confirm windowed counts.

Summary

Create a Kafka topic to hold streaming events.

Run a Kafka Streams app configured to group events into time windows.

Produce events to the input topic and consume windowed results from the output topic.

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