GroupBy and aggregation in Kafka - Interactive Code Practice

Practice - 5 Tasks

1fill in blank

easy

Complete the code to group the stream by the 'user' field.

Kafka

KGroupedStream<String, String> groupedStream = stream.[1]();

Drag options to blanks, or click blank then click option'

Amap

BflatMap

Cfilter

DgroupByKey

Attempts:

3 left

💡 Hint

Common Mistakes

Using filter instead of groupByKey

Using map which transforms records but does not group

Using flatMap which expands records instead of grouping

The groupByKey method groups the stream by its existing key, which is 'user' here.

2fill in blank

medium

Complete the code to count the number of records per key.

Kafka

KTable<String, Long> counts = groupedStream.[1]();

Drag options to blanks, or click blank then click option'

Aaggregate

Breduce

Ccount

DmapValues

Attempts:

3 left

💡 Hint

Common Mistakes

Using reduce which combines values but does not count

Using aggregate which requires more parameters

Using mapValues which transforms values but does not count

The count method counts the number of records for each key in the grouped stream.

3fill in blank

hard

Fix the error in the aggregation code by completing the missing method.

Kafka

KTable<String, Integer> aggregated = groupedStream.[1](() -> 0, (key, value, aggregate) -> aggregate + Integer.parseInt(value));

Drag options to blanks, or click blank then click option'

Aaggregate

Bmap

Creduce

Dcount

Attempts:

3 left

💡 Hint

Common Mistakes

Using count which does not take initializer or aggregator

Using reduce which does not take initializer

Using map which transforms but does not aggregate

The aggregate method allows custom aggregation with initializer and aggregator functions.

4fill in blank

hard

Fill both blanks to group the stream by 'category' and sum the values as integers.

Kafka

KTable<String, Integer> summed = stream.mapValues(Integer::parseInt).[1]( (key, value) -> key.split(":")[0] ).[2](Integer::sum);

Drag options to blanks, or click blank then click option'

AgroupBy

BmapValues

Creduce

Dcount

Attempts:

3 left

💡 Hint

Common Mistakes

Using mapValues instead of groupBy

Using count instead of reduce

Using reduce without grouping first

groupBy groups by a new key extracted from the original key. reduce sums the integer values.

5fill in blank

hard

Fill the blanks to create a grouped stream by 'type', aggregate values starting at 0, and add parsed integers.

Kafka

KTable<String, Integer> result = stream.[1]( (key, value) -> value.split(",")[0] ).[2](() -> 0, (aggKey, newValue, aggValue) -> aggValue + Integer.parseInt(newValue));

Drag options to blanks, or click blank then click option'

AgroupBy

Baggregate

CmapValues

Dcount

Attempts:

3 left

💡 Hint

Common Mistakes

Using count instead of aggregate

Using mapValues instead of aggregate

Not grouping before aggregating

groupBy groups by the first part of the value, aggregate initializes and updates the aggregation, and mapValues is not used here but included as a distractor.

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