Filter and map operations in Kafka - Commands & Configuration

Introduction

Kafka streams allow you to process data in real time by filtering unwanted messages and transforming data into a new form. This helps you focus only on the important information and change it as needed while it flows through your system.

When you want to keep only messages with specific values from a stream of data.

When you need to change the format or content of messages before sending them to another system.

When you want to remove noisy or irrelevant data from your processing pipeline.

When you want to enrich or simplify data by applying a function to each message.

When you want to build real-time dashboards that show only filtered and transformed data.

Config File - filter_map_streams.properties

filter_map_streams.properties

application.id=filter-map-app
bootstrap.servers=localhost:9092
key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde

This configuration file sets up the Kafka Streams application with a unique application ID and connects it to the Kafka server running locally. It also defines how keys and values in messages are converted to and from bytes using string serializers and deserializers.

Commands

Create an input topic where raw messages will be sent for processing.

Terminal

kafka-topics --create --topic input-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

Expected Output

Created topic input-topic.

→

--topic - Name of the Kafka topic to create

→

--partitions - Number of partitions for the topic

→

--replication-factor - Number of copies of the data for fault tolerance

Start a producer to send messages to the input topic. You can type messages here to test filtering and mapping.

Terminal

kafka-console-producer --topic input-topic --bootstrap-server localhost:9092

Expected Output

No output (command runs silently)

→

--topic - Topic to send messages to

→

--bootstrap-server - Kafka server address

Run the Kafka Streams application that filters messages containing the word 'error' and converts messages to uppercase.

Terminal

java -cp kafka-streams-examples.jar org.apache.kafka.streams.examples.filtermap.FilterMapExample filter_map_streams.properties

Expected Output

Starting FilterMapExample

Topology:

... (topology details) ...

Streams started

Consume messages from the output topic to see the filtered and transformed results.

Terminal

kafka-console-consumer --topic output-topic --bootstrap-server localhost:9092 --from-beginning

Expected Output

ERROR: DISK FAILURE

WARNING: CPU HIGH

INFO: SYSTEM OK

→

--topic - Topic to read messages from

→

--from-beginning - Read all messages from the start

Key Concept

If you remember nothing else from this pattern, remember: filtering removes unwanted messages and mapping changes message content as data flows through Kafka Streams.

Common Mistakes

Not setting the correct serializers and deserializers in the configuration.

Kafka Streams cannot convert messages properly, causing errors or data loss.

Always specify matching key and value serdes for your message types in the properties file.

Trying to filter or map on the wrong topic or without starting the streams application.

No processing happens, so output topic remains empty or unchanged.

Make sure to create and send messages to the input topic and run the streams app before consuming output.

Summary

Create an input Kafka topic to receive raw messages.

Run a Kafka Streams app configured to filter and map messages.

Send messages to the input topic and consume processed messages from the output topic.

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