Filter and map operations in Kafka - Mini Project: Build & Apply

Filter and map operations

📖 Scenario: You are working with a stream of messages representing orders in a Kafka topic. Each message contains an order ID and the order amount. You want to process this stream to find only the orders with amounts greater than 100 and then create a new stream with just the order IDs of these large orders.

🎯 Goal: Build a Kafka Streams application that filters orders with amounts greater than 100 and maps them to their order IDs.

📋 What You'll Learn

Create a KStream named orders with sample order data

Create a variable threshold set to 100

Use filter on orders to keep only orders with amount greater than threshold

Use map to transform filtered orders to just their order IDs

Print the resulting stream to the console

💡 Why This Matters

🌍 Real World

Filtering and mapping streams is common in real-time data processing, such as monitoring orders, transactions, or sensor data.

💼 Career

Kafka Streams skills are valuable for backend developers, data engineers, and anyone working with real-time data pipelines.

1

Create the initial orders stream

Create a KStream called orders with these exact entries: ("order1", 50), ("order2", 150), ("order3", 200), ("order4", 80).

Kafka

# Create a KStream named orders with the given order data
# Your code here

Hint

Use builder.stream("orders-topic") to create the stream. The actual data is assumed to be in the topic.

2

Set the threshold value

Create an integer variable called threshold and set it to 100.

Kafka

KStream orders = builder.stream("orders-topic");
// For this exercise, assume orders stream contains: ("order1", 50), ("order2", 150), ("order3", 200), ("order4", 80)

// Create threshold variable
# Your code here

Hint

Just create a simple integer variable named threshold and assign 100.

3

Filter and map the orders stream

Use filter on orders to keep only entries where the value (order amount) is greater than threshold. Then use map to transform the filtered stream to only the order IDs (the keys). Store the result in a KStream called largeOrderIds.

Kafka

KStream orders = builder.stream("orders-topic");
// For this exercise, assume orders stream contains: ("order1", 50), ("order2", 150), ("order3", 200), ("order4", 80)

int threshold = 100;

// Filter orders and map to order IDs
# Your code here

Hint

Use filter with a lambda checking amount > threshold. Then use map with a lambda returning KeyValue.pair(orderId, orderId).

4

Print the filtered order IDs

Use foreach on largeOrderIds to print each order ID with the exact format: "Large order ID: <orderId>".

Kafka

KStream orders = builder.stream("orders-topic");
// For this exercise, assume orders stream contains: ("order1", 50), ("order2", 150), ("order3", 200), ("order4", 80)

int threshold = 100;

KStream largeOrderIds = orders
    .filter((orderId, amount) -> amount > threshold)
    .map((orderId, amount) -> KeyValue.pair(orderId, orderId));

// Print each large order ID
# Your code here

Hint

Use foreach with a lambda that prints the string exactly as shown.

Practice

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

easy

A. Deletes all messages in the stream

B. Changes messages to a new form

C. Keeps only messages that meet a condition

D. Sends messages to a different topic

Solution

1. Step 1: Understand filter purpose

   The filter operation selects messages that satisfy a given condition, removing others.

2. Step 2: Compare with other operations

   Unlike map, which changes message content, filter only keeps or drops messages.

3. Final Answer:

   Keeps only messages that meet a condition -> Option C

4. Quick Check:

   Filter = Keep messages by condition [OK]

Hint: Filter keeps or drops messages based on a rule [OK]

Common Mistakes:

• Confusing filter with map operation
• Thinking filter changes message content
• Assuming filter deletes all messages

2. Which of the following is the correct syntax to apply a map operation on a Kafka stream named stream to convert messages to uppercase?

easy

A. stream.map(value.toUpperCase)

B. stream.filter(value -> value.toUpperCase())

C. stream.map(value => value.toUpperCase())

D. stream.map(value -> value.toUpperCase())

Solution

1. Step 1: Identify correct lambda syntax in Kafka Streams

   Kafka Streams in Java uses map(value -> ...) syntax for mapping values.

2. Step 2: Check method call correctness

   value.toUpperCase() correctly converts string to uppercase.

3. Final Answer:

   stream.map(value -> value.toUpperCase()) -> Option D

4. Quick Check:

   Correct lambda and method call = stream.map(value -> value.toUpperCase()) [OK]

Hint: Use arrow -> for lambda in Kafka Streams Java [OK]

Common Mistakes:

• Using filter instead of map
• Using JavaScript arrow => instead of Java ->
• Missing parentheses in method call

3. Given the following Kafka Streams code, what will be the output messages after processing?

stream.filter(value -> value.length() > 3)
      .map(value -> value.substring(0, 3))

Input messages: ["apple", "cat", "banana", "dog"]

medium

A. ["cat", "dog"]

B. ["app", "ban"]

C. ["apple", "banana"]

D. ["app", "cat", "ban", "dog"]

Solution

1. Step 1: Apply filter condition

   Keep messages with length > 3: "apple" (5), "banana" (6). "cat" and "dog" are dropped.

2. Step 2: Apply map operation

   Take first 3 characters: "apple" -> "app", "banana" -> "ban".

3. Final Answer:

   ["app", "ban"] -> Option B

4. Quick Check:

   Filter then map = ["app", "ban"] [OK]

Hint: Filter first, then map to transform filtered messages [OK]

Common Mistakes:

• Not filtering before mapping
• Mapping before filtering
• Including messages with length ≤ 3

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

stream.filter(value -> value.startsWith("a"))
      .map(value -> value.toUpperCase())
      .filter(value -> value.length() > 5)

medium

A. No error; code is correct

B. Using two filter operations is not allowed

C. The second filter uses the mapped uppercase value correctly

D. The lambda syntax is incorrect

Solution

1. Step 1: Check filter and map chaining

   Chaining multiple filters and maps is allowed in Kafka Streams.

2. Step 2: Verify lambda syntax and logic

   Lambdas use correct syntax and logic: first filter checks start, map uppercases, second filter checks length.

3. Final Answer:

   No error; code is correct -> Option A

4. Quick Check:

   Multiple filters and maps allowed = No error; code is correct [OK]

Hint: Multiple filters and maps can be chained without error [OK]

Common Mistakes:

• Thinking multiple filters are disallowed
• Confusing lambda syntax with other languages
• Assuming map changes variable type incorrectly

5. You want to process a Kafka stream of user names to keep only those starting with 'J' and convert them to lowercase. Which code snippet correctly does this?

hard

A. stream.filter(name -> name.startsWith("J"))
.map(name -> name.toLowerCase())

B. stream.map(name -> name.toLowerCase())
.filter(name -> name.startsWith("J"))

C. stream.filter(name -> name.toLowerCase().startsWith("j"))
.map(name -> name.toUpperCase())

D. stream.map(name -> name.startsWith("J"))
.filter(name -> name.toLowerCase())

Solution

1. Step 1: Filter names starting with 'J'

   Use filter(name -> name.startsWith("J")) to keep only names starting with uppercase 'J'.

2. Step 2: Map filtered names to lowercase

   Apply map(name -> name.toLowerCase()) to convert names to lowercase.

3. Final Answer:

   stream.filter(name -> name.startsWith("J")) .map(name -> name.toLowerCase()) -> Option A

4. Quick Check:

   Filter first, then map to lowercase [OK]

Hint: Filter first by condition, then map to transform [OK]

Common Mistakes:

• Mapping before filtering causing wrong results
• Using incorrect lambda syntax
• Filtering after changing case causing misses