Streaming joins in Apache Spark - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

Output of a stream-stream inner join with watermarks

Consider two streaming DataFrames df1 and df2 with event-time columns eventTime1 and eventTime2. Both have watermarks set with 10 minutes delay. They are joined on a key column and event times within 5 minutes. What is the output count after processing the following batches?

Batch 1:
df1: (key=1, eventTime1=10:00), (key=2, eventTime1=10:05)
df2: (key=1, eventTime2=10:03), (key=2, eventTime2=10:20)

Batch 2:
df1: (key=1, eventTime1=10:15)
df2: (key=1, eventTime2=10:10)

Apache Spark

from pyspark.sql.functions import expr

# Assume df1 and df2 are streaming DataFrames with schema (key int, eventTime timestamp)
# Watermarks set as:
df1_with_watermark = df1.withWatermark('eventTime1', '10 minutes')
df2_with_watermark = df2.withWatermark('eventTime2', '10 minutes')

joined = df1_with_watermark.alias("df1").join(
    df2_with_watermark.alias("df2"),
    expr("df1.key = df2.key AND df2.eventTime2 BETWEEN df1.eventTime1 AND df1.eventTime1 + interval 5 minutes"),
    'inner'
)

# After processing batches, count the joined rows
result_count = joined.count()

A 2

B 3

C 4

D 1

💡 Hint

Remember that watermarks drop late data and the join condition requires event times within 5 minutes.

In batch 1, only key=1 matches because 10:03 is within 10:00 to 10:05. Key=2 does not match because 10:20 is not within 10:05 to 10:10. In batch 2, 10:10 does not match the new df1 at 10:15 (10:10 < 10:15), nor previous df1 key=1 at 10:00 (10:10 > 10:05). So total matches are 1.

❓ data_output

intermediate

Number of rows after left outer streaming join with late data

You have two streaming DataFrames orders and customers. You perform a left outer join on customerId. The customers stream has a watermark of 1 hour on updateTime. If a late customer update arrives after the watermark, will it appear in the join output?

Given the following data:

Orders batch:
(orderId=101, customerId=1, orderTime=12:00)
(orderId=102, customerId=2, orderTime=12:05)

Customers batch:
(customerId=1, updateTime=11:00, name='Alice')
(customerId=2, updateTime=10:00, name='Bob')

A Both customers appear in join output

B Only customerId=2 appears, customerId=1 is dropped due to watermark

C Only customerId=1 appears, customerId=2 is dropped due to late update

D No customers appear because watermark drops all

💡 Hint

Watermark drops data older than current watermark threshold.

Customer with customerId=2 has updateTime=10:00 which is older than watermark threshold (assuming current time near 12:00). So this late data is dropped and does not join. CustomerId=1 is within watermark and joins successfully.

🔧 Debug

advanced

Identify the error in stream-stream join code

Given the following Spark Structured Streaming code snippet, what error will it raise?

joined = df1.join(df2, on='key', how='inner')
query = joined.writeStream.format('console').start()
query.awaitTermination()

Assume both df1 and df2 are streaming DataFrames without watermarks.

Apache Spark

joined = df1.join(df2, on='key', how='inner')
query = joined.writeStream.format('console').start()
query.awaitTermination()

A No error, code runs and outputs joined data

B AnalysisException: stream-stream join without watermark is not supported

C TimeoutException due to missing trigger interval

D NullPointerException due to missing schema

💡 Hint

Stream-stream joins require watermarks to handle state cleanup.

Spark requires watermarks on both streams for stream-stream joins to manage state and avoid infinite state growth. Without watermarks, it raises AnalysisException.

❓ visualization

advanced

Visualize state growth in stream-stream join with and without watermark

You run two streaming joins: one with watermarks on both streams and one without watermarks. Which graph best represents the state size over time?

A Graph showing state size growing linearly without watermark, stable with watermark

B Graph showing state size stable without watermark, growing with watermark

C Graph showing state size dropping to zero immediately in both cases

D Graph showing random fluctuations in state size regardless of watermark

💡 Hint

Watermarks help Spark remove old state data.

Without watermarks, Spark cannot remove old state, so state size grows linearly. With watermarks, old state is cleaned up, keeping state size stable.

🚀 Application

expert

Design a streaming join to handle late data with event-time tolerance

You need to join two streams clicks and impressions on userId with event-time columns clickTime and impressionTime. You want to join clicks with impressions that happened within 10 minutes before or after the click. Late data can arrive up to 30 minutes late. Which approach correctly implements this with Spark Structured Streaming?

A Set watermarks on both streams with 30 minutes delay, join on userId and condition: impressionTime BETWEEN clickTime - interval 10 minutes AND clickTime + interval 10 minutes

B Set watermark only on clicks with 10 minutes delay, join on userId and impressionTime = clickTime

C No watermarks, join on userId and impressionTime BETWEEN clickTime - interval 30 minutes AND clickTime + interval 30 minutes

D Set watermarks on both streams with 10 minutes delay, join on userId and impressionTime = clickTime

💡 Hint

Watermarks must cover maximum allowed lateness. Join condition must cover the time window.

To handle late data up to 30 minutes, watermarks must be set with 30 minutes delay on both streams. The join condition uses a time range of ±10 minutes around clickTime to match impressions. This ensures correct matches and state cleanup.

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