Executing Stream-to-Stream Joins

Perform inner or outer joins between two distinct streaming sources by defining time-based constraints to manage memory and state growth.

In this lesson, you will learn how to combine two continuous data streams using time-based constraints to maintain system stability and state management.

The Challenge of Stream-to-Stream Joins

When you join two static datasets, the engine can scan the entire tables. However, when joining two live streams (e.g., an Orders stream and a Shipments stream), the system doesn't know when a matching record might arrive. If you didn't define boundaries, the engine would have to store every single event from both sides in memory indefinitely to check for potential future matches. This is a recipe for memory exhaustion and system crashes.

Applying Time-Based Constraints

To perform a join between two streams, we define a Time-to-Live (TTL) or a window constraint. We tell the streaming engine, "Only keep this record in state for X amount of time while waiting for its counterpart."

For example, if an order must be shipped within 24 hours, there is no reason to keep the Order record in memory once that window has passed. By setting a constraint like join_key = join_key AND order_time BETWEEN shipment_time - INTERVAL 1 HOUR AND shipment_time + INTERVAL 24 HOURS, you explicitly tell the engine to clear its state for that specific key once the window expires.

Example Concept

Consider a user clicking an advertisement and then making a purchase. We want to join these two streams:

SELECT *
FROM AdClicks AS clicks
INNER JOIN Purchases AS buys
ON clicks.userId = buys.userId
AND buys.time BETWEEN clicks.time AND clicks.time + INTERVAL 30 MINUTES

In this case, the BETWEEN clause acts as a physical boundary. The engine stores AdClicks in its state store, but only for 30 minutes. Once an AdClick record is older than 30 minutes without finding a matching Purchase, the engine safely discards it from the state, preventing the memory from growing infinitely.

Inner vs. Outer Joins

Inner Joins: Only records that find a match within the time constraint are output.
Outer Joins: If no match is found within the time constraint, the stream outputs the record from the "preserved" side (e.g., the AdClick) with NULL values for the missing Purchase data, helping you identify unmatched events.

Key Takeaway: Stream-to-stream joins require explicit time-based constraints to prevent state bloat, ensuring the engine knows exactly how long to buffer records before discarding them from memory.