GroupBy and aggregation in Kafka - Commands & Configuration

Introduction

When you have many messages flowing through Kafka, you often want to group them by a key and calculate summaries like counts or sums. GroupBy and aggregation help you organize and analyze streaming data in real time.

When you want to count how many times each user sends messages in a stream.

When you need to sum sales amounts grouped by product category as data flows in.

When you want to find the average temperature per city from sensor data streaming through Kafka.

When you want to track the number of events per minute grouped by event type.

When you want to create a real-time leaderboard by grouping scores by player.

Config File - kafka-streams-aggregation.properties

kafka-streams-aggregation.properties

application.id=groupby-aggregation-app
bootstrap.servers=localhost:9092
cache.max.bytes.buffering=0
processing.guarantee=at_least_once
default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
default.value.serde=org.apache.kafka.common.serialization.Serdes$LongSerde

This configuration file sets up the Kafka Streams application with a unique application ID and connects it to the Kafka server at localhost:9092. It disables caching for immediate results and sets serialization for keys as strings and values as long integers, which are common for counting or summing operations.

Commands

Create a Kafka topic named 'sales' with 3 partitions to distribute data and 1 replication for fault tolerance.

Terminal

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

Expected Output

Created topic sales.

→

--partitions - Number of partitions for parallelism

→

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

Start a producer to send sales data messages to the 'sales' topic.

Terminal

kafka-console-producer --topic sales --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 application that groups sales by product and sums the amounts in real time.

Terminal

java -jar kafka-streams-aggregation.jar

Expected Output

Starting Kafka Streams application...

Topology built successfully.

Streams started.

Consume and display the aggregated sales results from the 'aggregated-sales' topic to verify the groupBy and sum operation.

Terminal

kafka-console-consumer --topic aggregated-sales --bootstrap-server localhost:9092 --from-beginning

Expected Output

productA	150

productB	200

productA	180

→

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

Key Concept

If you remember nothing else from this pattern, remember: grouping messages by key and applying aggregation functions lets you summarize streaming data in real time.

Common Mistakes

Not setting the correct key serializer and deserializer in the configuration.

Kafka Streams cannot group messages properly without correct key serialization, causing runtime errors or wrong grouping.

Always configure key serde (serializer/deserializer) to match the key data type, like StringSerde for string keys.

Forgetting to create the output topic before running the streams application.

The application fails to write aggregated results if the output topic does not exist.

Create all input and output topics with appropriate partitions before starting the streams app.

Using caching without understanding it delays output results.

Caching buffers results and delays aggregation output, confusing beginners expecting immediate results.

Set cache.max.bytes.buffering=0 in config to disable caching for learning and immediate output.

Summary

Create Kafka topics to hold input and output data streams.

Configure Kafka Streams with proper serializers and disable caching for immediate aggregation.

Run a Kafka Streams app that groups messages by key and applies aggregation like sum or count.

Use console consumer to verify the aggregated results in the output topic.

Practice

1. What does the groupBy operation do in Kafka Streams?

easy

A. It sends data to an external database.

B. It deletes records from the stream.

C. It organizes records by a new key to prepare for aggregation.

D. It filters records based on a condition.

Solution

1. Step 1: Understand the purpose of groupBy

   The groupBy operation changes the key of records to group them logically.

2. Step 2: Connect grouping to aggregation

   Grouping is necessary before aggregation to compute summaries per group.

3. Final Answer:

   It organizes records by a new key to prepare for aggregation. -> Option C

4. Quick Check:

   groupBy = organize by new key [OK]

Hint: GroupBy changes keys to group data before aggregation [OK]

Common Mistakes:

• Thinking groupBy deletes data
• Confusing groupBy with filtering
• Assuming groupBy sends data outside Kafka

2. Which of the following is the correct syntax to perform a count aggregation after grouping a Kafka stream named orders by customerId?

easy

A. orders.count().groupBy(value -> value.customerId);

B. orders.groupBy((key, value) -> value.customerId).count();

C. orders.groupByKey().sum();

D. orders.aggregate((key, value) -> value.customerId).count();

Solution

1. Step 1: Identify correct groupBy syntax

   The groupBy method takes a key selector function like (key, value) -> value.customerId.

2. Step 2: Apply count aggregation after grouping

   Calling count() after groupBy correctly counts records per group.

3. Final Answer:

   orders.groupBy((key, value) -> value.customerId).count(); -> Option B

4. Quick Check:

   groupBy then count = correct syntax [OK]

Hint: Group first, then aggregate like count() [OK]

Common Mistakes:

• Calling count() before groupBy
• Using groupByKey() without key selector
• Mixing aggregate() with wrong parameters

3. Given the following Kafka Streams code snippet, what is the output type of customerCounts?

var customerCounts = orders
  .groupBy((key, order) -> order.customerId)
  .count();

medium

A. KTable<String, Long>

B. KStream<String, Integer>

C. KStream<String, Long>

D. KTable<String, String>

Solution

1. Step 1: Understand groupBy and count return types

   Grouping returns a grouped stream, and count() returns a KTable with counts per key.

2. Step 2: Identify key and value types

   The key is String (customerId), and count returns Long values.

3. Final Answer:

   KTable<String, Long> -> Option A

4. Quick Check:

   count() returns KTable with Long values [OK]

Hint: count() after groupBy returns KTable with Long values [OK]

Common Mistakes:

• Assuming count returns KStream
• Confusing Long with Integer type
• Mixing up KStream and KTable

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

var result = orders
  .groupBy(order -> order.customerId)
  .count();

medium

A. Missing a filter() before groupBy.

B. count() cannot be called after groupBy.

C. The variable 'orders' must be a KTable, not a KStream.

D. groupBy requires a key-value mapper, not a single argument.

Solution

1. Step 1: Check groupBy method signature

   groupBy expects a function with two parameters: key and value, e.g., (key, value) -> newKey.

2. Step 2: Identify incorrect lambda usage

   The code uses a single-argument lambda order -> order.customerId, which is invalid for groupBy.

3. Final Answer:

   groupBy requires a key-value mapper, not a single argument. -> Option D

4. Quick Check:

   groupBy needs (key, value) -> newKey [OK]

Hint: groupBy needs two parameters in lambda [OK]

Common Mistakes:

• Using single-argument lambda for groupBy
• Thinking count() is invalid after groupBy
• Assuming orders must be KTable

5. You want to compute the total sales amount per product category from a Kafka stream sales where each record has category and amount. Which Kafka Streams code snippet correctly performs this aggregation?

hard

A. sales.groupBy((key, sale) -> sale.category).aggregate(() -> 0.0, (key, newVal, agg) -> agg + newVal.amount);

B. sales.groupBy((key, sale) -> sale.category).reduce((agg, newVal) -> agg + newVal.amount);

C. sales.groupByKey().count();

D. sales.countBy((key, sale) -> sale.category);

Solution

1. Step 1: Choose aggregation method for summing amounts

   To sum amounts, use aggregate() with initializer and adder function.

2. Step 2: Verify groupBy key selector and aggregation logic

   Grouping by sale.category and adding newVal.amount to accumulator is correct.

3. Final Answer:

   sales.groupBy((key, sale) -> sale.category).aggregate(() -> 0.0, (key, newVal, agg) -> agg + newVal.amount); -> Option A

4. Quick Check:

   aggregate with initializer sums amounts per category [OK]

Hint: Use aggregate() with initializer to sum values per group [OK]

Common Mistakes:

• Using reduce() without matching types
• Calling count() instead of sum
• Using non-existent countBy() method