How do I do a stream-stream join?

Summary

• A stream-stream join in Spark Structured Streaming (added in Spark 2.3) buffers past input from both streams as state, so every new row can be matched against everything seen so far.
• Inner joins work on any columns, but state grows without bound unless you set watermarks on both inputs and add an event-time constraint to the join condition.
• Outer and semi joins must have a watermark and an event-time constraint to produce correct results.
• The programming guide supports joins in Append output mode only, and Spark 4.2.0 adds Update mode for non-outer stream-stream joins.
• With several watermarks, Spark uses the minimum as the global watermark by default.

How do I do a stream-stream join?

To join two streaming DataFrames, put a watermark on each input and write a join condition that includes an event-time range. The watermarks and the range tell Spark when buffered rows can no longer match, so it can drop them from state. Without them, Spark keeps all past input, because any row from one stream can match a future row from the other.

What does a stream-stream join look like?

This example follows the ad impressions and clicks case from the Spark guide, with rate sources standing in for real streams:

from pyspark.sql.functions import expr

impressions = (spark.readStream.format("rate").option("rowsPerSecond", 5).load()
    .selectExpr("value AS impressionAdId", "timestamp AS impressionTime"))
clicks = (spark.readStream.format("rate").option("rowsPerSecond", 5).load()
    .selectExpr("value AS clickAdId", "timestamp AS clickTime"))

# Apply watermarks on event-time columns
impressionsWithWatermark = impressions.withWatermark("impressionTime", "2 hours")
clicksWithWatermark = clicks.withWatermark("clickTime", "3 hours")

# Join with event-time constraints
joined = impressionsWithWatermark.join(
    clicksWithWatermark,
    expr("""
        clickAdId = impressionAdId AND
        clickTime >= impressionTime AND
        clickTime <= impressionTime + interval 1 hour
    """))

query = joined.writeStream.format("console").start()

The watermark delays say impressions and clicks can arrive up to 2 and 3 hours late in event time. The range condition says a click occurs within 0 seconds to 1 hour after its impression. You can define the constraint as a time range condition or as a join on event-time windows. With a 2 hour delay, data inside the delay is never dropped, and data delayed by more may or may not be processed.

Which join types are supported?

Pass the join type as the how argument of join, for example inner, left_outer, right_outer, full_outer, or left_semi.

Stream-stream join	Support	Watermark and time constraints
Inner	Supported	Optional on both sides, needed to clean up state
Left outer	Conditionally supported	Watermark on the right input required, left optional
Right outer	Conditionally supported	Watermark on the left input required, right optional
Full outer	Conditionally supported	Watermark on one side required, other optional
Left semi	Conditionally supported	Watermark on the right input required, left optional

Outer and semi joins also need event-time constraints for correct results, because the engine must know when a row will never match. A stream joined with a static DataFrame is a different case: stream-static joins are not stateful, so they need no state management.

What output mode do joins use?

The programming guide says joins work only in Append mode, and its output-mode matrix lists Append alone for join queries. The Spark 4.2.0 release notes add "Support stream-stream non-outer join in Update mode" (SPARK-56384). Outer results arrive late: NULL rows wait for the watermark delay and the time range, and watermarks advance at the end of each micro-batch. If either input stops receiving data, outer output can be delayed.

How do multiple watermarks interact?

Spark tracks the maximum event time of each input and picks one global watermark. The default is the minimum, so the watermark moves at the pace of the slowest stream. Setting spark.sql.streaming.multipleWatermarkPolicy to max gets faster results but aggressively drops data from slower streams. You cannot change that setting between restarts from the same checkpoint.

What else should I know?

• Joins can be cascaded, as in df1.join(df2, ...).join(df3, ...).
• You cannot use mapGroupsWithState or flatMapGroupsWithState before and after joins.
• Between restarts, changes to a join's schema, equi-joining columns, or join type are not allowed.
• Stream-stream joins initialize four state store instances per partition, according to the Databricks monitoring docs.

How does Databricks handle stream-stream joins?

Databricks runs the same Structured Streaming API and documents stream-stream joins under stateful streaming. Databricks Runtime 13.3 LTS and above can chain multiple stateful operators, such as a windowed aggregation feeding a join, in append output mode only. Databricks recommends RocksDB with changelog checkpointing for stateful queries.
In Lakeflow Pipelines, a stream-stream join must have a watermark on both sides and a time interval clause that uses the same fields as the watermarks. The pipeline SQL from the Databricks docs looks like this:

CREATE OR REFRESH STREAMING TABLE
  silver.adImpressionClicks
AS SELECT
  imp.userId, impressionAdId, clickTimestamp, impressionSeconds
FROM STREAM
  (bronze.rawAdImpressions)
WATERMARK
  impressionTimestamp DELAY OF INTERVAL 3 MINUTES imp
INNER JOIN STREAM
  (bronze.rawClicks)
WATERMARK clickTimestamp DELAY OF INTERVAL 3 MINUTES click
ON
  imp.userId = click.userId
AND
  clickAdId = impressionAdId
AND
  clickTimestamp >= impressionTimestamp
AND
  clickTimestamp <= impressionTimestamp + interval 3 minutes

FAQs

Can I join more than two streams?

Yes. Cascade the joins, so one join's result feeds the next.

Do I need a watermark for an inner join?

It is optional for correctness, but without it all past input stays in state and the state keeps growing.

What happens if the watermark is too short?

Rows later than the threshold may be dropped. The Databricks docs note that a smaller threshold emits records sooner, but late records are more likely to be dropped.