Streaming joins in Apache Spark - Mini Project: Build & Apply

Streaming Joins with Apache Spark

📖 Scenario: You work at a company that processes live data streams from two sources: user clicks and user profiles. You want to join these streams to enrich click data with user information in real time.

🎯 Goal: Build a Spark Structured Streaming application that reads two streaming DataFrames, performs a join on user ID, and outputs the enriched click data.

📋 What You'll Learn

Create streaming DataFrames for clicks and user profiles

Set a watermark on event time columns to handle late data

Perform an inner join on user ID between the two streams

Write the joined stream to the console sink

💡 Why This Matters

🌍 Real World

Streaming joins are used in real-time analytics to combine live data from multiple sources, such as user activity and profile information, to provide enriched insights instantly.

💼 Career

Data engineers and data scientists use streaming joins to build pipelines that process and analyze live data for monitoring, personalization, and alerting systems.

1

Create streaming DataFrames for clicks and user profiles

Create two streaming DataFrames called clicks and profiles by reading from JSON files in directories "/path/to/clicks" and "/path/to/profiles" respectively. Use spark.readStream with schema option set to the given schemas. The clicks schema has fields userId (string) and clickTime (timestamp). The profiles schema has fields userId (string) and userName (string).

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, TimestampType

spark = SparkSession.builder.appName("StreamingJoins").getOrCreate()

clicks_schema = StructType([
    StructField("userId", StringType()),
    StructField("clickTime", TimestampType())
])

profiles_schema = StructType([
    StructField("userId", StringType()),
    StructField("userName", StringType())
])

# Create streaming DataFrame clicks reading from "/path/to/clicks" with clicks_schema
# Create streaming DataFrame profiles reading from "/path/to/profiles" with profiles_schema
# Your code here

Hint

Use spark.readStream.schema(...).json(path) to create streaming DataFrames.

2

Set watermark on event time columns

Add watermarks to the streaming DataFrames to handle late data. Set a watermark of 10 minutes on clicks using the clickTime column, and set a watermark of 10 minutes on profiles using the userId column (even though it is not a timestamp, just follow the instruction for practice). Assign the results back to clicks and profiles.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, TimestampType

spark = SparkSession.builder.appName("StreamingJoins").getOrCreate()

clicks_schema = StructType([
    StructField("userId", StringType()),
    StructField("clickTime", TimestampType())
])

profiles_schema = StructType([
    StructField("userId", StringType()),
    StructField("userName", StringType())
])

clicks = spark.readStream.schema(clicks_schema).json("/path/to/clicks")
profiles = spark.readStream.schema(profiles_schema).json("/path/to/profiles")

# Add watermark of 10 minutes on clicks using clickTime
# Add watermark of 10 minutes on profiles using userId
# Your code here

Hint

Use withWatermark(column, "10 minutes") on each DataFrame.

3

Perform inner join on userId

Use an inner join to combine clicks and profiles on the userId column. Assign the joined DataFrame to a variable called joined_stream.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, TimestampType

spark = SparkSession.builder.appName("StreamingJoins").getOrCreate()

clicks_schema = StructType([
    StructField("userId", StringType()),
    StructField("clickTime", TimestampType())
])

profiles_schema = StructType([
    StructField("userId", StringType()),
    StructField("userName", StringType())
])

clicks = spark.readStream.schema(clicks_schema).json("/path/to/clicks")
profiles = spark.readStream.schema(profiles_schema).json("/path/to/profiles")

clicks = clicks.withWatermark("clickTime", "10 minutes")
profiles = profiles.withWatermark("userId", "10 minutes")

# Join clicks and profiles on userId using inner join
# Assign the result to joined_stream
# Your code here

Hint

Use clicks.join(profiles, on="userId", how="inner") to join the streams.

4

Write the joined stream to console

Write the joined_stream to the console sink using writeStream. Set the output mode to append and start the query. Assign the query to a variable called query. Finally, print "Streaming join started".

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, TimestampType

spark = SparkSession.builder.appName("StreamingJoins").getOrCreate()

clicks_schema = StructType([
    StructField("userId", StringType()),
    StructField("clickTime", TimestampType())
])

profiles_schema = StructType([
    StructField("userId", StringType()),
    StructField("userName", StringType())
])

clicks = spark.readStream.schema(clicks_schema).json("/path/to/clicks")
profiles = spark.readStream.schema(profiles_schema).json("/path/to/profiles")

clicks = clicks.withWatermark("clickTime", "10 minutes")
profiles = profiles.withWatermark("userId", "10 minutes")

joined_stream = clicks.join(profiles, on="userId", how="inner")

# Write joined_stream to console with outputMode 'append' and start the query
# Assign the query to variable query
# Print "Streaming join started"
# Your code here

Hint

Use writeStream.format("console").outputMode("append").start() and assign to query.

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