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'

Need a 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'

Need a 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

Need a 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

Need a hint?

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