GroupBy and aggregation in Kafka - Step-by-Step Execution

Process Flow - GroupBy and aggregation

Input Stream of Records

↓

Group records by key

↓

Apply aggregation function

↓

Emit aggregated results

↓

Output Stream

Data flows in, records are grouped by a key, then an aggregation function combines values per group, producing summarized output.

Execution Sample

Kafka

KStream<String, Integer> stream = builder.stream("input-topic");
KTable<String, Integer> aggregated = stream
  .groupByKey()
  .reduce((v1, v2) -> v1 + v2);
aggregated.toStream().to("output-topic");

This code groups records by key and sums their integer values, then writes results to an output topic.

Process Table

Step	Input Record (key,value)	GroupBy Key	Aggregation Function	Aggregated Value	Output Emitted
1	(A, 5)	A	Initial value 5	5	Yes (A,5)
2	(B, 3)	B	Initial value 3	3	Yes (B,3)
3	(A, 7)	A	5 + 7	12	Yes (A,12)
4	(B, 2)	B	3 + 2	5	Yes (B,5)
5	(A, 1)	A	12 + 1	13	Yes (A,13)
6	No more input	-	-	-	Stream ends

💡 No more input records, aggregation completes.

Status Tracker

Variable	Start	After 1	After 2	After 3	After 4	After 5	Final
Aggregated Value for A	null	5	5	12	12	13	13
Aggregated Value for B	null	null	3	3	5	5	5

Key Moments - 3 Insights

Why is there output after the first record?

The first record sets the initial aggregated value (without applying the aggregation function) and emits it. Further records apply the function and emit updates.

How does the aggregation function combine values?

It adds the new value to the current aggregated value for the key, as shown in steps 3, 4, and 5 in the execution table.

What happens when no more input records arrive?

The stream ends and no further aggregation updates occur, as noted in the exit note and step 6.

Visual Quiz - 3 Questions

Look at the execution table, what is the aggregated value for key 'A' after step 3?

A 5

B 7

C 12

D 13

💡 Hint

Check the 'Aggregated Value' column at step 3 for key 'A'.

At which step does the output for key 'B' first get emitted?

A Step 4

B Step 2

C Step 3

D Step 5

💡 Hint

Look at the 'Output Emitted' column for key 'B' in the execution table.

If the aggregation function was changed to multiply values instead of add, what would be the aggregated value for key 'A' after step 5?

A 105

B 35

C 65

D 13

💡 Hint

Multiply values 5 * 7 * 1 for key 'A' from steps 1, 3, and 5.

Concept Snapshot

GroupBy and aggregation in Kafka Streams:
- Use groupByKey() to group records by key.
- Apply aggregation like reduce() to combine values per key.
- Aggregation updates on each new record.
- Results emitted as a KTable stream.
- Useful for counting, summing, or other summaries.

Full Transcript

This visual trace shows how Kafka Streams processes records by grouping them by key and aggregating their values. Each input record updates the aggregated value for its key using the aggregation function. Outputs are emitted when aggregation changes. The process stops when no more input records arrive.

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