GroupBy and aggregation in Kafka - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

What is the output of this Kafka Streams aggregation?

Consider the following Kafka Streams code snippet that groups records by key and counts the number of records per key.

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> stream = builder.stream("input-topic");
KTable<String, Long> counts = stream
    .groupByKey()
    .count();

counts.toStream().foreach((key, value) -> System.out.println(key + ":" + value));

If the input topic contains the records with keys: a, b, a, a, b, what will be printed?

Kafka

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> stream = builder.stream("input-topic");
KTable<String, Long> counts = stream
    .groupByKey()
    .count();

counts.toStream().foreach((key, value) -> System.out.println(key + ":" + value));

A a:3\nb:2

B a:2\nb:3

C a:1\nb:1

D a:3\nb:3

💡 Hint

Count counts how many records exist per key.

The count() aggregation counts the number of records per key. Since the input keys are 'a', 'b', 'a', 'a', 'b', the counts are 'a' = 3 and 'b' = 2.

❓ Predict Output

intermediate

What is the output of this Kafka Streams reduce operation?

Given the following Kafka Streams code that groups records by key and reduces values by concatenation:

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> stream = builder.stream("input-topic");
KTable<String, String> reduced = stream
    .groupByKey()
    .reduce((v1, v2) -> v1 + v2);

reduced.toStream().foreach((key, value) -> System.out.println(key + ":" + value));

If the input topic contains records with keys and values: (a, x), (b, y), (a, z), (a, w), (b, q), what will be printed?

Kafka

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> stream = builder.stream("input-topic");
KTable<String, String> reduced = stream
    .groupByKey()
    .reduce((v1, v2) -> v1 + v2);

reduced.toStream().foreach((key, value) -> System.out.println(key + ":" + value));

A a:xzw\nb:qy

B a:xzw\nb:yq

C a:xz\nb:yq

D a:zwx\nb:yq

💡 Hint

Reduce concatenates values in the order they arrive per key.

The reduce function concatenates values for each key in the order they arrive. For key 'a', values are 'x', 'z', 'w' concatenated as 'xzw'. For key 'b', values are 'y', 'q' concatenated as 'yq'.

❓ Predict Output

advanced

What is the output of this Kafka Streams aggregate with initializer?

Consider this Kafka Streams aggregation that sums integer values per key:

StreamsBuilder builder = new StreamsBuilder();
KStream<String, Integer> stream = builder.stream("input-topic", Consumed.with(Serdes.String(), Serdes.Integer()));
KTable<String, Integer> aggregated = stream
    .groupByKey()
    .aggregate(
        () -> 0,
        (key, value, aggregate) -> aggregate + value
    );

aggregated.toStream().foreach((key, value) -> System.out.println(key + ":" + value));

If the input topic contains records: (a, 2), (b, 3), (a, 5), (a, -1), (b, 4), what will be printed?

Kafka

StreamsBuilder builder = new StreamsBuilder();
KStream<String, Integer> stream = builder.stream("input-topic", Consumed.with(Serdes.String(), Serdes.Integer()));
KTable<String, Integer> aggregated = stream
    .groupByKey()
    .aggregate(
        () -> 0,
        (key, value, aggregate) -> aggregate + value
    );

aggregated.toStream().foreach((key, value) -> System.out.println(key + ":" + value));

A a:6\nb:7

B a:7\nb:7

C a:5\nb:7

D a:6\nb:4

💡 Hint

Aggregate sums values starting from 0 per key.

The aggregate starts at 0 for each key. For 'a': 0 + 2 + 5 + (-1) = 6. For 'b': 0 + 3 + 4 = 7.

❓ Predict Output

advanced

What error does this Kafka Streams code raise?

Examine this Kafka Streams code snippet:

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> stream = builder.stream("input-topic");
KTable<String, Long> counts = stream
    .groupBy((key, value) -> value)
    .count();

What error will this code raise when compiled or run?

Kafka

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> stream = builder.stream("input-topic");
KTable<String, Long> counts = stream
    .groupBy((key, value) -> value)
    .count();

A Runtime error: NullPointerException due to missing key serializer

B No error, code runs correctly

C Compilation error: count() requires a Materialized instance

D Compilation error: groupBy requires a KeyValueMapper but lambda returns String only

💡 Hint

Check the expected input type for groupBy.

The groupBy method expects a KeyValueMapper<K, V, KR> that returns a key type. The lambda returns only the value, which is a String, but the method signature expects a KeyValueMapper that returns a key type. This mismatch causes a compilation error.

🧠 Conceptual

expert

How many items are in the resulting KTable after this aggregation?

Given a Kafka Streams KStream with records having keys: a, b, c, a, b, a and values: 1, 2, 3, 4, 5, 6, the following aggregation is performed:

KTable<String, Integer> result = stream
    .groupByKey()
    .aggregate(
        () -> 0,
        (key, value, aggregate) -> aggregate + value
    );

After processing all records, how many entries will the result KTable contain?

A 1

B 6

C 3

D 0

💡 Hint

Count distinct keys after grouping.

The aggregation groups by key. The keys are 'a', 'b', and 'c'. So the resulting KTable will have 3 entries, one per distinct key.