Streaming joins in Apache Spark - Step-by-Step Execution

Concept Flow - Streaming joins

Start Streaming DataFrames

↓

Define Join Condition

↓

Apply Streaming Join

↓

Process Joined Stream

↓

Output Results Continuously

↓

End

Streaming joins combine two continuous data streams based on a condition, producing joined output as new data arrives.

Execution Sample

Apache Spark

stream1 = spark.readStream.format("socket").option("host", "localhost").option("port", 9998).load()
stream2 = spark.readStream.format("socket").option("host", "localhost").option("port", 9999).load()
joined = stream1.join(stream2, stream1.value == stream2.value)
query = joined.writeStream.format("console").start()

This code reads two streaming sources from sockets, joins them on matching values, and outputs the joined data to the console continuously.

Execution Table

Step	Action	Input Data	Join Condition	Joined Output
1	Read first batch from stream1	["apple", "banana"]	N/A	N/A
2	Read first batch from stream2	["banana", "cherry"]	N/A	N/A
3	Apply join on stream1.value == stream2.value	stream1: ["apple", "banana"]; stream2: ["banana", "cherry"]	value equality	["banana"]
4	Output joined data to console	["banana"]	value equality	["banana"]
5	Read second batch from stream1	["date"]	N/A	N/A
6	Read second batch from stream2	["date", "apple"]	N/A	N/A
7	Apply join on new batches	stream1: ["date"]; stream2: ["date", "apple"]	value equality	["date"]
8	Output joined data to console	["date"]	value equality	["date"]
9	No more data, streaming continues waiting	N/A	N/A	N/A

💡 Streaming continues indefinitely; here we show two batches processed and output.

Variable Tracker

Variable	Start	After Step 1	After Step 2	After Step 3	After Step 5	After Step 6	After Step 7	Final
stream1_data	[]	["apple", "banana"]	["apple", "banana"]	["apple", "banana"]	["date"]	["date"]	["date"]	["date"]
stream2_data	[]	[]	["banana", "cherry"]	["banana", "cherry"]	["banana", "cherry"]	["date", "apple"]	["date", "apple"]	["date", "apple"]
joined_output	[]	[]	[]	["banana"]	["banana"]	["banana"]	["date"]	["date"]

Key Moments - 2 Insights

Why does the join output only show matching values from both streams?

Because the join condition requires values to be equal in both streams, only those matching values appear in the joined output as shown in execution_table rows 3 and 7.

What happens if one stream has data but the other does not?

No joined output is produced until matching data arrives in both streams, as seen in steps where one stream has new data but the other does not, no join output is emitted.

Visual Quiz - 3 Questions

Look at the execution table at step 3. What is the joined output?

A ["banana"]

B ["apple"]

C ["cherry"]

D []

💡 Hint

Check the 'Joined Output' column at step 3 in the execution_table.

At which step does the join output include the value "date"?

A Step 5

B Step 3

C Step 7

D Step 9

💡 Hint

Look at the 'Joined Output' column for the value "date" in the execution_table.

If stream2 never sends "banana", what happens to the joined output at step 3?

A It outputs ["banana"] anyway

B It outputs an empty list []

C It outputs all values from stream1

D It stops the stream

💡 Hint

Refer to the join condition and output behavior in execution_table rows 3 and 4.

Concept Snapshot

Streaming joins combine two live data streams by matching rows based on a condition.
They continuously output joined results as new data arrives.
Use Spark's readStream to read, join() to combine, and writeStream to output.
Only matching data from both streams appears in output.
Streaming joins run indefinitely until stopped.

Full Transcript

Streaming joins in Apache Spark combine two continuous data streams based on a join condition, such as matching values. The process starts by reading streaming data from two sources. Then, a join condition is defined, for example, matching the 'value' field in both streams. The join is applied continuously as new data arrives. Joined results are output continuously, for example, to the console. The execution table shows step-by-step how batches from each stream are read, joined, and output. Variables track the data in each stream and the joined output after each step. Key moments clarify why only matching values appear in the output and what happens if one stream lacks matching data. The visual quiz tests understanding of the joined output at specific steps and the effect of missing data. The snapshot summarizes the key points: streaming joins combine live data streams, output matching rows continuously, and require both streams to have matching data to produce output.

Practice

1. What is the main purpose of a streaming join in Apache Spark?

easy

A. To combine two live data streams based on matching keys

B. To save streaming data permanently to disk

C. To convert batch data into streaming data

D. To filter out duplicate records in a stream

Solution

1. Step 1: Understand streaming join concept

   Streaming joins are used to combine two live streams by matching keys in real time.

2. Step 2: Differentiate from other operations

   Saving data, converting batch to stream, or filtering duplicates are different tasks, not streaming joins.

3. Final Answer:

   To combine two live data streams based on matching keys -> Option A

4. Quick Check:

   Streaming join = combine streams by keys [OK]

Hint: Streaming joins combine live streams on keys [OK]

Common Mistakes:

• Confusing streaming join with saving data
• Thinking streaming join converts batch to stream
• Mixing up filtering duplicates with joining

2. Which of the following is the correct syntax to perform a streaming join between two DataFrames streamA and streamB on column id in Spark?

easy

A. streamA.concat(streamB, 'id')

B. streamA.merge(streamB, on='id')

C. streamA.join(streamB, streamA.id == streamB.id)

D. streamA.union(streamB).filter('id')

Solution

1. Step 1: Recall Spark join syntax

   Spark uses .join() with a join condition like streamA.id == streamB.id.

2. Step 2: Check other options

   .merge(), .concat(), and .union() are not valid for streaming joins in Spark.

3. Final Answer:

   streamA.join(streamB, streamA.id == streamB.id) -> Option C

4. Quick Check:

   Streaming join syntax = .join(condition) [OK]

Hint: Use .join() with condition for streaming joins [OK]

Common Mistakes:

• Using .merge() instead of .join()
• Trying to concatenate streams for join
• Using union which just stacks data

3. Given the following Spark Structured Streaming code, what will be the output schema of the joined stream?

streamA = spark.readStream.format('socket').option('host', 'localhost').option('port', 9999).load()
streamB = spark.readStream.format('socket').option('host', 'localhost').option('port', 9998).load()
joinedStream = streamA.join(streamB, streamA.value == streamB.value)
joinedStream.printSchema()

medium

A. StructType with only one 'value' column

B. StructType with two 'value' columns from streamA and streamB

C. Error due to ambiguous column names

D. Empty schema because streams are empty

Solution

1. Step 1: Understand join output schema

   Joining two streams with same column names results in both columns included with source prefixes.

2. Step 2: Check for ambiguity handling

   Spark automatically prefixes columns to avoid ambiguity, so no error occurs.

3. Final Answer:

   StructType with two 'value' columns from streamA and streamB -> Option B

4. Quick Check:

   Join output = both columns with prefixes [OK]

Hint: Joined streams keep both columns with prefixes [OK]

Common Mistakes:

• Assuming only one column remains after join
• Expecting error due to same column names
• Thinking schema is empty before data arrives

4. Consider this code snippet for joining two streaming DataFrames:

joinedStream = streamA.join(streamB, 'id')

When running this code, you get a runtime error about missing watermark. What is the likely cause?

medium

A. The join type is missing

B. The join condition 'id' is invalid syntax

C. The streams have different schemas

D. No watermark defined on one or both streams causing state to grow indefinitely

Solution

1. Step 1: Understand watermark role in streaming joins

   Watermarks limit how long Spark keeps state for late data in streaming joins.

2. Step 2: Identify cause of runtime error

   Without watermark, Spark cannot clean old state, causing errors about state size.

3. Final Answer:

   No watermark defined on one or both streams causing state to grow indefinitely -> Option D

4. Quick Check:

   Missing watermark = runtime error in streaming join [OK]

Hint: Always set watermark to avoid state errors [OK]

Common Mistakes:

• Thinking join condition syntax is wrong
• Assuming schema mismatch causes this error
• Believing join type must always be specified

5. You have two streaming DataFrames: orders with columns orderId, customerId, orderTime and customers with columns customerId, customerName, updateTime. You want to join them on customerId but only match orders with customer info updated within 1 hour before the order. Which approach correctly implements this streaming join?

hard

A. Use event-time watermark on both streams and join with condition: orders.customerId = customers.customerId AND customers.updateTime BETWEEN orders.orderTime - interval 1 hour AND orders.orderTime

B. Join on customerId only without watermark or time condition

C. Use batch join after collecting streaming data

D. Join on customerId and filter results after join for time difference

Solution

1. Step 1: Use watermarks to handle late data

   Set watermarks on orderTime and updateTime to manage state and late events.

2. Step 2: Apply join condition with time range

   Join on customerId and add time condition to match updates within 1 hour before order.

3. Step 3: Avoid filtering after join or batch join

   Filtering after join wastes resources; batch join loses streaming benefits.

4. Final Answer:

   Use event-time watermark on both streams and join with condition: orders.customerId = customers.customerId AND customers.updateTime BETWEEN orders.orderTime - interval 1 hour AND orders.orderTime -> Option A

5. Quick Check:

   Watermark + time condition in join = correct streaming join [OK]

Hint: Use watermark and time range in join condition [OK]

Common Mistakes:

• Skipping watermark causing state issues
• Joining without time condition losing accuracy
• Filtering after join instead of in join condition
• Using batch join losing streaming benefits