Search⌘ K
AI Features
Home
Courses
Mastering Apache Kafka

The DSL API Stateful Operations: Join

Explore how to implement various join operations in Kafka Streams, such as KStream-KStream, KTable-KTable, foreign-key joins, and joins with GlobalKTables. Understand how to enrich streaming data with reference tables using these stateful operations and build real-time data processing applications.

We'll cover the following...

Join operations combine data from two or more input streams and/or tables based on a common key. Kafka Streams provides several types of join operations, including inner, left, and outer join, each of which has its own unique characteristics and use cases. These are used to combine related data streams and perform complex analyses and transformations on the resulting data. For example, joins can be used to combine clickstream data with customer data to create a more complete picture of customer behavior or combine data from different sensors in an IoT system to identify patterns and anomalies.

Types of join operations

Let’s explore the different types of join operations in Kafka Streams.

The KStream to KStream join

A KStream to KStream join combines two streams by matching records that share the same key and whose timestamps fall within a specified time window. The join is defined by a ValueJoiner, a function that takes the two matching record values and returns a single combined output value.

Let’s say we have two KStreams: one contains information about user activity and another about user location. We want to join these two streams based on the user ID to create a stream of enriched user activity events.

Java
KStream<String, String> userActivityStream = builder.stream("activity");
KStream<String, String> userLocationStream = builder.stream("location");
KStream<String, String> enrichedStream = userActivityStream.join(
    userLocationStream,
    (activity, location) -> "User " + activity + " was seen at " + location,
    JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5))
);

In the example above, we join the user activity stream with the user location stream based on the user ID key.

  • Line 6: The ValueJoiner lambda takes the activity and location values from the two streams and returns a string representing the enriched event.

  • Line 7: The JoinWindows argument sets the maximum time difference allowed between two records with matching keys for them to be joined (5 minutes here). ... ...