GroupBy and aggregation in Kafka - Mini Project: Build & Apply

Kafka Stream Processing: GroupBy and Aggregation

📖 Scenario: You are working with a Kafka stream that receives sales data from a store. Each message contains the product name and the quantity sold. You want to group the sales by product and calculate the total quantity sold for each product.

🎯 Goal: Build a Kafka Streams application that groups sales by product and aggregates the total quantity sold per product.

📋 What You'll Learn

Create a Kafka Streams topology with a source topic named sales.

Group the stream by product key.

Aggregate the total quantity sold per product.

Print the aggregated results to the console.

💡 Why This Matters

🌍 Real World

Retail companies use Kafka Streams to analyze live sales data by product to track inventory and demand in real time.

💼 Career

Understanding Kafka Streams grouping and aggregation is essential for roles in data engineering, real-time analytics, and DevOps managing streaming data pipelines.

1

Create the initial Kafka Streams source stream

Create a Kafka Streams KStream named salesStream that reads from the topic "sales" using the builder.stream("sales") method.

Kafka

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;

public class SalesAggregator {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();

        // Create salesStream from topic "sales"
        // Your code here
    }
}

Hint

Use builder.stream("sales") to create the stream and assign it to salesStream.

2

Parse the sales data and create a key-value pair

Create a new KStream named parsedStream by mapping salesStream to extract the product as key and quantity as integer value. Assume each value is a comma-separated string like "product,quantity". Use map with lambda (key, value) -> new KeyValue<>(product, quantity).

Kafka

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.KeyValue;

public class SalesAggregator {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();

        KStream salesStream = builder.stream("sales");

        // Map salesStream to parsedStream with product as key and quantity as Integer value
        // Your code here
    }
}

Hint

Split the value by comma, parse quantity to Integer, and return new KeyValue with product and quantity.

3

Group by product and aggregate total quantity

Create a KTable<String, Integer> named totalByProduct by grouping parsedStream by key using groupByKey() and aggregating with reduce to sum quantities. Use (aggValue, newValue) -> aggValue + newValue as the reducer.

Kafka

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.KeyValue;

public class SalesAggregator {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();

        KStream salesStream = builder.stream("sales");

        KStream parsedStream = salesStream.map((key, value) -> {
            String[] parts = value.split(",");
            String product = parts[0];
            Integer quantity = Integer.parseInt(parts[1]);
            return new KeyValue(product, quantity);
        });

        // Group parsedStream by key and aggregate total quantity
        // Your code here
    }
}

Hint

Use groupByKey() on parsedStream and then reduce to sum the quantities.

4

Print the aggregated results to the console

Use totalByProduct.toStream().foreach to print each product and its total quantity in the format "Product: <product>, Total Quantity: <quantity>". Then start the Kafka Streams application with KafkaStreams streams = new KafkaStreams(builder.build(), props); streams.start();. Assume props is already defined.

Kafka

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.KeyValue;
import java.util.Properties;

public class SalesAggregator {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("application.id", "sales-aggregator");
        props.put("bootstrap.servers", "localhost:9092");
        props.put("default.key.serde", Serdes.String().getClass().getName());
        props.put("default.value.serde", Serdes.String().getClass().getName());

        StreamsBuilder builder = new StreamsBuilder();

        KStream salesStream = builder.stream("sales");

        KStream parsedStream = salesStream.map((key, value) -> {
            String[] parts = value.split(",");
            String product = parts[0];
            Integer quantity = Integer.parseInt(parts[1]);
            return new KeyValue(product, quantity);
        });

        KTable totalByProduct = parsedStream.groupByKey().reduce(
            (aggValue, newValue) -> aggValue + newValue
        );

        // Print aggregated results and start streams
        // Your code here
    }
}

Hint

Use toStream().foreach on totalByProduct to print results, then create and start KafkaStreams with builder.build() and props.

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