Join operations (KStream-KStream, KStream-KTable) in Kafka - Mini Project: Build & Apply

Join operations (KStream-KStream, KStream-KTable)

📖 Scenario: You are working with Apache Kafka Streams to process real-time data. You have two streams of data: one with user clicks and another with user purchases. You want to join these streams to analyze user behavior.

🎯 Goal: Build a Kafka Streams application that performs a KStream-KStream join and a KStream-KTable join to combine user click and purchase data.

📋 What You'll Learn

Create a KStream named clicksStream with user click events

Create a KStream named purchasesStream with user purchase events

Create a KTable named usersTable with user profile data

Perform a KStream-KStream join between clicksStream and purchasesStream

Perform a KStream-KTable join between clicksStream and usersTable

Print the results of both joins

💡 Why This Matters

🌍 Real World

Joining streams and tables is common in real-time analytics to combine different data sources for richer insights.

💼 Career

Kafka Streams joins are essential skills for data engineers and DevOps professionals working with event-driven architectures.

1

Create initial KStreams for clicks and purchases

Create two KStreams named clicksStream and purchasesStream from the topics user-clicks and user-purchases respectively using the builder.stream() method.

Kafka

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;

StreamsBuilder builder = new StreamsBuilder();
// Create clicksStream from topic 'user-clicks'
// Create purchasesStream from topic 'user-purchases'

Hint

Use builder.stream("topic-name") to create a KStream from a Kafka topic.

2

Create a KTable for user profiles

Create a KTable named usersTable from the topic user-profiles using the builder.table() method.

Kafka

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;

StreamsBuilder builder = new StreamsBuilder();

KStream clicksStream = builder.stream("user-clicks");
KStream purchasesStream = builder.stream("user-purchases");
// Create usersTable from topic 'user-profiles'

Hint

Use builder.table("topic-name") to create a KTable from a Kafka topic.

3

Perform KStream-KStream and KStream-KTable joins

Perform a KStream-KStream join between clicksStream and purchasesStream using join() with a window of 5 minutes. Then perform a KStream-KTable join between clicksStream and usersTable using leftJoin(). Store the results in joinedClicksPurchases and joinedClicksUsers respectively.

Kafka

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.JoinWindows;
import java.time.Duration;

StreamsBuilder builder = new StreamsBuilder();

KStream clicksStream = builder.stream("user-clicks");
KStream purchasesStream = builder.stream("user-purchases");

KTable usersTable = builder.table("user-profiles");

// Perform KStream-KStream join with 5 minutes window
// Perform KStream-KTable left join

Hint

Use join() with JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)) for KStream-KStream join. Use leftJoin() for KStream-KTable join.

4

Print the joined streams

Print the contents of joinedClicksPurchases and joinedClicksUsers streams using foreach() with a lambda that prints the key and value.

Kafka

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.JoinWindows;
import java.time.Duration;

StreamsBuilder builder = new StreamsBuilder();

KStream clicksStream = builder.stream("user-clicks");
KStream purchasesStream = builder.stream("user-purchases");

KTable usersTable = builder.table("user-profiles");

KStream joinedClicksPurchases = clicksStream.join(purchasesStream,
    (click, purchase) -> "Click: " + click + ", Purchase: " + purchase,
    JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)));

KStream joinedClicksUsers = clicksStream.leftJoin(usersTable,
    (click, user) -> "Click: " + click + ", User: " + (user == null ? "unknown" : user));

// Print joinedClicksPurchases
// Print joinedClicksUsers

Hint

Use foreach((key, value) -> System.out.println(...)) to print each record in the stream.

Practice

1. What is the main difference between a KStream-KStream join and a KStream-KTable join in Kafka Streams?

easy

A. KStream-KStream joins require a time window; KStream-KTable joins do not.

B. KStream-KTable joins require a time window; KStream-KStream joins do not.

C. Both joins require a time window to match records.

D. Neither join requires a time window.

Solution

1. Step 1: Understand KStream-KStream join behavior

   KStream-KStream joins combine records from two streams that occur within a specified time window.

2. Step 2: Understand KStream-KTable join behavior

   KStream-KTable joins enrich stream records with the latest matching table record without needing a time window.

3. Final Answer:

   KStream-KStream joins require a time window; KStream-KTable joins do not. -> Option A

4. Quick Check:

   Time window needed only for KStream-KStream joins [OK]

Hint: Remember: only KStream-KStream joins need time windows [OK]

Common Mistakes:

• Confusing which join requires a time window
• Thinking both joins behave the same
• Assuming KStream-KTable join needs time window

2. Which of the following is the correct syntax to perform a KStream-KTable join in Kafka Streams?

easy

A. ktable.join(kstream, (tableValue, streamValue) -> tableValue + streamValue);

B. kstream.join(ktable, JoinWindows.of(Duration.ofMinutes(5)), (v1, v2) -> v1 + v2);

C. kstream.join(ktable, (streamValue, tableValue) -> streamValue + tableValue);

D. kstream.join(ktable, (v1, v2) -> v1 + v2, JoinWindows.of(Duration.ofMinutes(5)));

Solution

1. Step 1: Identify KStream-KTable join syntax

   KStream-KTable join uses kstream.join(ktable, ValueJoiner) without a time window.

2. Step 2: Check options for correct method signature

   kstream.join(ktable, (streamValue, tableValue) -> streamValue + tableValue); matches the correct syntax: join without window and with a ValueJoiner lambda.

3. Final Answer:

   kstream.join(ktable, (streamValue, tableValue) -> streamValue + tableValue); -> Option C

4. Quick Check:

   KStream-KTable join syntax = kstream.join(ktable, (streamValue, tableValue) -> streamValue + tableValue); [OK]

Hint: KStream-KTable join has no time window parameter [OK]

Common Mistakes:

• Adding JoinWindows to KStream-KTable join
• Reversing stream and table parameters
• Using KTable join method on KStream

3. Given the following Kafka Streams code snippet, what will be the output if the input streams have matching keys within the 5-minute window?

var joinedStream = stream1.join(stream2, (v1, v2) -> v1 + v2, JoinWindows.of(Duration.ofMinutes(5)));
joinedStream.foreach((key, value) -> System.out.println(key + ":" + value));

medium

A. Prints combined values for keys appearing in both streams within 5 minutes

B. Prints combined values for all keys in stream1 regardless of stream2

C. Prints only keys from stream2 with no combination

D. Throws a runtime exception due to missing KTable

Solution

1. Step 1: Understand KStream-KStream join with window

   The join combines records from both streams with matching keys occurring within the 5-minute window.

2. Step 2: Analyze output behavior

   For matching keys within the window, the values are combined using the lambda (v1 + v2) and printed.

3. Final Answer:

   Prints combined values for keys appearing in both streams within 5 minutes -> Option A

4. Quick Check:

   KStream-KStream join outputs combined matching keys [OK]

Hint: KStream-KStream join outputs combined values within window [OK]

Common Mistakes:

• Assuming all keys from stream1 print regardless of stream2
• Confusing KStream-KTable join behavior
• Expecting runtime error without KTable

4. Identify the error in the following Kafka Streams join code:

var joined = stream1.join(stream2, (v1, v2) -> v1 + v2);
joined.print();

medium

A. Incorrect lambda syntax for join function

B. Missing JoinWindows parameter for KStream-KStream join

C. Using print() instead of foreach() causes error

D. stream2 should be a KTable, not a KStream

Solution

1. Step 1: Check join method parameters

   KStream-KStream join requires a JoinWindows parameter to specify the time window for matching events.

2. Step 2: Identify missing parameter

   The code omits the JoinWindows argument, causing a compilation error.

3. Final Answer:

   Missing JoinWindows parameter for KStream-KStream join -> Option B

4. Quick Check:

   KStream-KStream join needs JoinWindows [OK]

Hint: KStream-KStream join always needs JoinWindows [OK]

Common Mistakes:

• Forgetting JoinWindows in KStream-KStream join
• Confusing KStream-KTable join syntax
• Assuming print() is invalid method

5. You want to enrich a stream of user clicks (KStream) with the latest user profile data stored in a KTable. Which join operation should you use and why?

hard

A. Use KStream-KStream join without a window to join clicks and profiles.

B. Use KStream-KStream join with a 10-minute window to combine clicks and profiles.

C. Use KTable-KTable join because both are tables.

D. Use KStream-KTable join to enrich clicks with latest profile data without time window.

Solution

1. Step 1: Identify data types and join needs

   User clicks are a stream (KStream), and user profiles are stored as a table (KTable).

2. Step 2: Choose appropriate join type

   KStream-KTable join enriches stream data with the latest table data for matching keys without needing a time window.

3. Final Answer:

   Use KStream-KTable join to enrich clicks with latest profile data without time window. -> Option D

4. Quick Check:

   KStream-KTable join enriches stream with latest table data [OK]

Hint: Enrich stream with table data using KStream-KTable join [OK]

Common Mistakes:

• Using KStream-KStream join unnecessarily
• Adding time window to KStream-KTable join
• Confusing KTable-KTable join for stream enrichment