advanced
Stream Joins
10 min readLast updated: 2026-07-09
Overview
Learn how to join streaming data with static datasets (Stream-to-Static) or combine two real-time streams (Stream-to-Stream).
What You Will Learn
In this lesson, you will learn:
- Stream-to-Static Join: Enriching streams using lookup tables.
- Stream-to-Stream Join: Combining two real-time events.
- Watermark Joins: Preventing infinite state storage during stream joins.
Detailed Concept Explanation
Spark Structured Streaming supports joining streaming datasets.
Join Types
- Stream-to-Static Joins: Joins a streaming DataFrame with a static DataFrame. This is stateless and does not require watermarks. The static table is read once (or cached) and joined as stream rows arrive.
- Stream-to-Stream Joins: Joins two streaming DataFrames. This requires Spark to buffer events in memory (state store) to find matches. To prevent memory leaks, you must configure Watermarks and time-bound constraints on both streams.
Code Examples
Input Dataset Preview
Below is the static lookup table of user departments:
| userId | dept |
|---|---|
| 1 | IT |
Python (PySpark) Implementation
python
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("StreamJoins").getOrCreate()
# Create static lookup DataFrame
static_users = spark.createDataFrame([(1, "IT")], ["userId", "dept"])
# Stream user events
events_stream = spark.readStream.format("rate").load() \
.selectExpr("CAST(value AS INT) AS userId", "timestamp")
# Stream-to-Static Join
joined_stream = events_stream.join(static_users, "userId")
query = joined_stream.writeStream.format("console").start()
query.awaitTermination(5)
query.stop()
Expected Output
text
+------+-------------------+----+
|userId| timestamp|dept|
+------+-------------------+----+
| 1|2026-07-09 19:30:00| IT|
+------+-------------------+----+
Execution Plan Diagram (Python & Scala)
Execution Plan Diagram
SparkSession.builder
createDataFrame(static)
readStream(events)
join(static_users
userId)
writeStream
start()
Scala Implementation
scala
import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder().appName("StreamJoinsScala").getOrCreate()
import spark.implicits._
val staticUsers = Seq((1, "IT")).toDF("userId", "dept")
val eventsStream = spark.readStream.format("rate").load()
.selectExpr("CAST(value AS INT) AS userId", "timestamp")
val joined = eventsStream.join(staticUsers, "userId")
val query = joined.writeStream.format("console").start()
query.awaitTermination(5000)
query.stop()
Expected Output
text
+------+-------------------+----+
|userId| timestamp|dept|
+------+-------------------+----+
| 1|2026-07-09 19:30:00| IT|
+------+-------------------+----+
Common Mistakes
- Failing to Define Watermarks for Stream-to-Stream Joins: Attempting to join two streams without setting watermarks and time-bound constraints. Spark will keep all events in memory forever to check for matches, which will eventually crash the executors.