Skip to main content
Mapagam

Implementing Stream Joins

1. Understanding Join Types

Join Type Description Use Case Example
Stream-Stream Join Joins two Kafka streams based on a common key. Real-time data aggregation from multiple sources. 
KStream<String, String> joined = stream1.join(stream2,
    (value1, value2) -> value1 + " " + value2,
    JoinWindows.of(Duration.ofMinutes(5)),
    StreamJoined.with(Serdes.String(), Serdes.String(), Serdes.String())
);
Stream-Table Join Joins a stream with a Kafka table ( changelog ) based on key. Enriching stream data with static or slowly changing data. 
KStream<String, String> enriched = stream.join(table,
    (streamValue, tableValue) -> streamValue + " " + tableValue,
    JoinWindows.of(Duration.ofMinutes(5))
);
Table-Table Join Joins two Kafka tables (changelog) on a key. Combining static data sources for analytics. 
KTable<String, String> joined = table1.join(table2,
    (value1, value2) -> value1 + " " + value2
);
Note: Join windows define the time window in which records can be joined.

2. Using Stream-Stream Join

A stream-stream join combines two Kafka streams based on a common key and allows for processing of data from both streams in a single operation.

Property Description Values
join.window.ms Time window for joining streams 30000
left.stream Left input stream name user-events
right.stream Right input stream name user-profiles
join.key Key used for joining streams user_id

Example: Stream-Stream Join Configuration

const kafka = new Kafka({ brokers: ['localhost:9092'] });
const consumer = kafka.consumer();
const producer = kafka.producer();

// Define the streams
const userEventsStream = consumer.subscribe({ topic: 'user-events' });
const userProfilesStream = consumer.subscribe({ topic: 'user-profiles' });

// Perform the join operation
const joinedStream = Streams.join(
  userEventsStream,
  userProfilesStream,
  (event, profile) => {
    return { ...event, ...profile };
  },
  { window: 30000 }
);

// Process joined data
joinedStream.forEach((message) => {
  console.log('Joined message:', message);
});
Note: The join window determines how long to wait for matching records from both streams before considering them as unmatched.

3. Using Stream-Table Join

A stream-table join combines a continuous stream of events with a changelog table (e.g., a configuration or lookup table) to produce a new stream of results. This is useful for enriching streaming data with static or slowly-changing reference data.

Property Description Example
stream The continuous stream to join with the table ordersStream
table The changelog table to join with the stream customerTable
joinKey The field used to match records between stream and table customerId
outputStream The resulting stream after the join operation enrichedOrdersStream

Example: Stream-Table Join in Kafka Streams

StreamsBuilder builder = new StreamsBuilder();
KStream<String, Order> ordersStream = builder.stream("orders");
KTable<String, Customer> customerTable = builder.table("customers");

KStream<String, EnrichedOrder> enrichedOrders = ordersStream.join(
    customerTable,
    (orderId, order) -> order.getCustomerId(),
    (order, customer) -> new EnrichedOrder(order, customer)
);

enrichedOrders.to("enriched-orders");
Note: The joinKey must be present in both the stream and table records to perform a successful join.
Warning: Stream-table joins are stateful and require a changelog topic for the table. Ensure that the table topic is properly configured to support updates.

4. Using Table-Table Join

A table-table join in Kafka Streams allows you to join two Kafka topics that represent tables (i.e., they contain key-value records where keys are unique). This type of join is useful for enriching data from one topic with data from another.

Property Description Values
joinType Specifies the type of join to perform INNER, LEFT, RIGHT, FULL
table1 Name of the first Kafka topic (table) orders
table2 Name of the second Kafka topic (table) customers
keyField Field used for joining records customerId

Example: Table-Table Join

KTable<String, Order> orders = builder.table("orders");
KTable<String, Customer> customers = builder.table("customers");

KTable<String, OrderWithCustomer> joined = orders.join(
    customers,
    (order, customer) -> new OrderWithCustomer(order, customer),
    JoinWindows.of(Duration.ofHours(1)),
    Joined.with(Serdes.String(), new OrderSerde(), new CustomerSerde())
);
Note: The join operation requires both tables to be keyed by the same field for matching records.
Warning: Ensure that the join window is appropriately configured to avoid missing records due to timing differences.

Key Considerations

  • Both tables must have a common key field for joining.
  • Join windows should be set to accommodate data latency and out-of-order events.
  • Use appropriate Serdes for serialization/deserialization of keys and values.

5. Implementing Inner Join

An inner join in Kafka Streams combines records from two streams based on a common key, emitting only those records that have matching keys in both streams.

Property Description Example
join.window.ms Time window in milliseconds for joining records. 30000
left.stream Stream to join from the left side. user-events
right.stream Stream to join from the right side. user-profiles

Example: Inner Join Implementation

StreamsBuilder builder = new StreamsBuilder();
KStream<String, UserEvent> userEvents = builder.stream("user-events");
KStream<String, UserProfile> userProfiles = builder.stream("user-profiles");

KStream<String, JoinedUser> joinedStream = userEvents.join(
    userProfiles,
    (event, profile) -> new JoinedUser(event.getUserId(), event.getEvent(), profile.getName()),
    JoinWindows.of(Duration.ofMinutes(5)),
    StreamJoined.with(Serdes.String(), new UserEventSerde(), new UserProfileSerde())
);

joinedStream.to("joined-output");
Note: The inner join requires both streams to have records with matching keys within the specified window.

6. Implementing Left Join

Property Description Default Value Notes
join.window.ms Time window for joining records from left and right streams 60000 Milliseconds
left.stream Name of the left stream in join operation null Required field
right.stream Name of the right stream in join operation null Required field

Example: Left Join Configuration

const leftStream = streams.left;
const rightStream = streams.right;

const joinedStream = KStream.leftJoin(
  leftStream,
  rightStream,
  (leftValue, rightValue) => {
    return {
      id: leftValue.id,
      leftData: leftValue.data,
      rightData: rightValue.data
    };
  },
  JoinWindows.of(TimeWindows.of(60000))
);
Note: Left joins preserve all records from the left stream, even if there is no matching record in the right stream.

7. Implementing Outer Join

An outer join in Kafka Streams combines records from two streams based on a common key, including all records from both streams even if there's no match in the other stream. This is useful for scenarios where you want to preserve all data, even if some records don't have corresponding entries in the other stream.

Property Description Example
outerJoin Method to perform an outer join between two KStream objects kstream1.outerJoin(kstream2, ...)
joinWindow Defines the time window for joining records JoinWindows.of(TimeWindows.of(Duration.ofMinutes(5)))
leftJoin Performs a left outer join, preserving all records from the left stream kstream1.leftJoin(kstream2, ...)
rightJoin Performs a right outer join, preserving all records from the right stream kstream1.rightJoin(kstream2, ...)

Example: Performing an Outer Join

KStream<String, String> leftStream = builder.stream("left-topic");
KStream<String, String> rightStream = builder.stream("right-topic");

KStream<String, String> joinedStream = leftStream.outerJoin(
    rightStream,
    (leftValue, rightValue) -> leftValue + "|" + rightValue,
    JoinWindows.of(TimeWindows.of(Duration.ofMinutes(5))),
    StreamJoined.with(Serdes.String(), Serdes.String(), Serdes.String())
);
Note: The outer join operation will emit a record for each key in either stream, even if one side has no matching record. This can lead to a higher volume of output compared to inner joins.

8. Configuring Join Windows

Property Default Value Description
join.window.ms 60000 Join window duration in milliseconds. Controls how long to wait for matching records from the other stream.
join.window.start 0 Start time of the join window in milliseconds. Used for windowed joins with fixed windows.
join.window.end 0 End time of the join window in milliseconds. Used for windowed joins with fixed windows.

Example: Setting Join Window Duration

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> stream1 = builder.stream("topic1");
KStream<String, String> stream2 = builder.stream("topic2");

KStream<String, String> joinedStream = stream1.join(stream2,
    (value1, value2) -> value1 + " - " + value2,
    JoinWindows.of(Duration.ofMinutes(5))
);
Note: The join window duration must be sufficient to allow for out-of-order events and ensure that all relevant records are included in the join operation.

9. Setting Join Retention

Property Default Value Description
retention.ms 604800000 Retention period for join results in milliseconds (7 days)
retention.bytes -1 Maximum size of join results in bytes (-1 = unlimited)

Example: Configuring Join Retention

{
  "retention.ms": 1209600000,
  "retention.bytes": 1073741824
}
Note: Join retention settings apply to the internal topic used for storing join results. Adjust these based on your data processing window and storage constraints.

10. Understanding Join Semantics

Join Type Description Use Case Notes
Stream-Stream Join Joins two Kafka streams based on a common key. Real-time event processing with correlated data sources. Requires both streams to be keyed by the same field.
Stream-Table Join Joins a stream with a changelog table (KTable) based on key. Enriching stream events with static or slowly changing data. Supports left joins and inner joins; updates in table propagate to joined stream.
Table-Table Join Joins two changelog tables (KTables) on a shared key. Combining static data sources for real-time analytics. Supports inner joins and left joins; updates in either table propagate to result.

Example: Stream-Stream Join

KStream<String, Order> orders = builder.stream("orders");
KStream<String, Customer> customers = builder.stream("customers");

KStream<String, OrderCustomer> joined = orders.join(customers,
    (order, customer) -> new OrderCustomer(order, customer),
    JoinWindows.of(Duration.ofHours(1)),
    StreamJoined.with(Serdes.String(), orderSerde, customerSerde));

joined.to("order-customer-joins");
Note: Join windows define the time range in which records can be joined. Records outside this window are either dropped or handled according to the join strategy.

11. Optimizing Join Performance

Configuration Value Description
join.window.ms 60000 Window size for stream joins in milliseconds
cache.max.bytes.buffering 1048576 Maximum number of bytes to buffer per partition
max.task.idle.ms 0 Maximum idle time for a task before it's considered stale

Example: Configuring Join Window Size

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "join-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.JOIN_WINDOW_MS_CONFIG, 30000L); // 30 seconds
Note: Setting join.window.ms too low can cause data loss due to late-arriving records.
Warning: Large join.window.ms values can increase memory usage and latency.

Performance Tips

  • Use join.window.ms to balance between data completeness and latency
  • Monitor cache.max.bytes.buffering to prevent out-of-memory errors
  • Consider using KStream<K, V>.join() with appropriate windowing

Was this article helpful?

Related Cheat Sheets