Why stream processing transforms data in Kafka - See It in Action

Why Stream Processing Transforms Data

📖 Scenario: Imagine you run a small online store. You get many orders every minute. You want to quickly know which products are popular right now. To do this, you use stream processing to transform the live order data as it flows in.

🎯 Goal: Build a simple Kafka stream processing program that reads order data, transforms it to count products sold, and outputs the counts.

📋 What You'll Learn

Create a Kafka topic with sample order data

Set up a stream processing configuration variable

Write a stream processing logic to transform orders into product counts

Print the final product counts

💡 Why This Matters

🌍 Real World

Online stores and many apps use stream processing to analyze live data instantly, like counting popular products or tracking user actions.

💼 Career

Understanding stream processing basics helps in roles like data engineer, backend developer, or any job working with real-time data pipelines.

1

Create the Kafka topic with sample order data

Create a Kafka topic called orders with these exact messages: {"order_id": 1, "product": "apple"}, {"order_id": 2, "product": "banana"}, {"order_id": 3, "product": "apple"}.

Kafka

# Your code here to create the 'orders' topic and add messages

Hint

Use a list called orders with dictionaries for each order.

2

Set up the stream processing configuration

Create a variable called product_counts and set it to an empty dictionary {} to hold counts of each product.

Kafka

orders = [
    {"order_id": 1, "product": "apple"},
    {"order_id": 2, "product": "banana"},
    {"order_id": 3, "product": "apple"}
]
# Your code here to create product_counts dictionary

Hint

Use an empty dictionary to store counts.

3

Transform orders to count products

Use a for loop with variable order to iterate over orders. Inside the loop, get the product name from order["product"]. Increase the count for that product in product_counts by 1. If the product is not yet in product_counts, add it with count 1.

Kafka

orders = [
    {"order_id": 1, "product": "apple"},
    {"order_id": 2, "product": "banana"},
    {"order_id": 3, "product": "apple"}
]
product_counts = {}
# Your code here to count products from orders

Hint

Use a loop and check if product is already counted.

4

Print the product counts

Write print(product_counts) to display the final counts of products sold.

Kafka

orders = [
    {"order_id": 1, "product": "apple"},
    {"order_id": 2, "product": "banana"},
    {"order_id": 3, "product": "apple"}
]
product_counts = {}
for order in orders:
    product = order["product"]
    if product in product_counts:
        product_counts[product] += 1
    else:
        product_counts[product] = 1
# Your code here to print product_counts

Hint

Use print to show the dictionary with counts.

Practice

1. Why do we use stream processing to transform data in Kafka Streams?

easy

A. To make data more useful and meaningful in real time

B. To store data permanently without changes

C. To slow down data flow for analysis later

D. To delete unnecessary data immediately

Solution

1. Step 1: Understand the purpose of stream processing

   Stream processing changes data as it flows to add value or meaning instantly.

2. Step 2: Identify the correct purpose from options

   Only To make data more useful and meaningful in real time says it makes data more useful in real time, which matches the purpose.

3. Final Answer:

   To make data more useful and meaningful in real time -> Option A

4. Quick Check:

   Stream processing transforms data = Make data useful [OK]

Hint: Stream processing changes data to add value instantly [OK]

Common Mistakes:

• Thinking stream processing only stores data
• Believing it slows down data flow
• Confusing transformation with deletion

2. Which of the following is the correct way to apply a map transformation in Kafka Streams using Java?

easy

A. stream.reduce((v1, v2) -> v1 + v2)

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

C. stream.mapValues(value -> value.toUpperCase())

D. stream.collect(Collectors.toList())

Solution

1. Step 1: Identify the map transformation syntax

   In Kafka Streams, mapValues applies a function to each value, e.g., converting to uppercase.

2. Step 2: Match the correct syntax

   stream.mapValues(value -> value.toUpperCase()) uses mapValues with a lambda toUpperCase(), which is correct.

3. Final Answer:

   stream.mapValues(value -> value.toUpperCase()) -> Option C

4. Quick Check:

   map transformation syntax = mapValues with lambda [OK]

Hint: Map uses mapValues with a function on values [OK]

Common Mistakes:

• Using filter instead of map
• Using reduce for mapping
• Using collect which is not a stream operation here

3. Given the following Kafka Streams code snippet, what will be the output if the input stream has values ["apple", "banana", "cherry"]?

stream.filter((k, v) -> v.startsWith("b")).toList()

medium

A. ["banana"]

B. ["apple", "banana", "cherry"]

C. ["cherry"]

D. []

Solution

1. Step 1: Understand the filter condition

   The filter keeps only values starting with 'b'. From the list, only "banana" starts with 'b'.

2. Step 2: Determine the output list

   After filtering, the list contains only ["banana"].

3. Final Answer:

   ["banana"] -> Option A

4. Quick Check:

   Filter by startsWith('b') = ["banana"] [OK]

Hint: Filter keeps only matching values [OK]

Common Mistakes:

• Including all values without filtering
• Choosing values starting with wrong letters
• Confusing filter with map

4. Identify the error in this Kafka Streams code snippet that tries to filter out empty strings:

stream.filter((k, v) -> v != "")

medium

A. filter() cannot be used on streams

B. Missing semicolon at the end

C. Lambda syntax is incorrect

D. Using '!=' to compare strings instead of .equals()

Solution

1. Step 1: Check string comparison in Java

   In Java, '!=' compares object references, not string content. Use .equals() to compare strings.

2. Step 2: Identify the correct fix

   Replace 'value != ""' with '!value.equals("")' to properly filter out empty strings.

3. Final Answer:

   Using '!=' to compare strings instead of .equals() -> Option D

4. Quick Check:

   String comparison needs .equals() [OK]

Hint: Use .equals() to compare strings in Java [OK]

Common Mistakes:

• Using '==' or '!=' for string content comparison
• Assuming filter() is invalid on streams
• Ignoring lambda syntax correctness

5. You want to enrich a Kafka stream of user IDs by adding their names from a lookup table. Which approach best uses stream processing transformations?

hard

A. Reduce the stream to count user IDs

B. Use a join operation between the user ID stream and the user info table

C. Map the stream values to uppercase user IDs

D. Filter the stream to only user IDs present in the table

Solution

1. Step 1: Understand enrichment in stream processing

   Enrichment means adding extra data, here user names, to existing stream data.

2. Step 2: Identify the correct transformation

   Joining the user ID stream with the user info table adds the names, which is enrichment.

3. Final Answer:

   Use a join operation between the user ID stream and the user info table -> Option B

4. Quick Check:

   Enrich data = join streams/tables [OK]

Hint: Enrich data by joining streams with lookup tables [OK]

Common Mistakes:

• Using filter which only removes data
• Mapping to uppercase does not add info
• Reducing counts data, not enriches