Streaming joins in Apache Spark - Deep Dive

Overview - Streaming joins

What is it?

Streaming joins are a way to combine two continuous streams of data based on matching keys or conditions. Instead of joining static tables, streaming joins work on data that keeps arriving over time. This allows real-time analysis by linking related events from different sources as they happen. It is commonly used in systems that need instant insights from live data.

Why it matters

Without streaming joins, it would be very hard to connect related live data points quickly, such as matching user clicks with ad impressions or linking sensor readings from different devices in real time. This would slow down decision-making and reduce the value of streaming data. Streaming joins enable fast, continuous correlation of data, making real-time monitoring, alerting, and analytics possible.

Where it fits

Learners should first understand batch joins and basic streaming concepts like data streams and windows. After mastering streaming joins, they can explore advanced stream processing topics like state management, watermarking, and event-time processing. Streaming joins build on core Spark Structured Streaming knowledge and lead into complex real-time data pipelines.

Mental Model

Core Idea

Streaming joins continuously match and combine related records from two live data streams as new data arrives over time.

Think of it like...

Imagine two conveyor belts carrying different colored balls. Streaming joins are like a worker who picks matching colored balls from both belts as they pass by and pairs them together instantly.

Stream A ──┐       ┌── Joined Output
           │ Join │
Stream B ──┘       └── Continuous matching of records based on keys

Build-Up - 7 Steps

1

Foundation

Understanding basic data streams

Concept: Learn what a data stream is and how streaming data differs from static data.

A data stream is a continuous flow of data records arriving over time, like live tweets or sensor readings. Unlike static data stored in files or databases, streams keep updating and growing. Spark Structured Streaming treats streams as unbounded tables that grow with new rows.

Result

You can now recognize streaming data as ongoing, never-ending input rather than fixed datasets.

Understanding that streaming data is continuous and unbounded is key to grasping why streaming joins need special handling compared to batch joins.

2

Foundation

Recap of batch joins in Spark

Concept: Review how joins work on static datasets before applying them to streams.

In batch processing, joins combine two static tables by matching rows with the same key. For example, joining a customer table with an orders table on customer ID. Spark reads both tables fully, then outputs the combined result.

Result

You understand the basic join operation as a way to combine related data from two fixed tables.

Knowing batch joins helps you see what changes when data is no longer static but streaming.

3

Intermediate

Introducing streaming joins concept

🤔Before reading on: Do you think streaming joins wait for all data before joining, or join as data arrives? Commit to your answer.

Concept: Streaming joins combine two live streams continuously, matching new records as they arrive without waiting for all data.

Unlike batch joins, streaming joins must handle data arriving at different times and possibly out of order. Spark supports joining two streams or a stream with a static dataset. The join happens incrementally, producing output as matching records appear.

Result

You see streaming joins as ongoing, incremental matching processes rather than one-time operations.

Understanding that streaming joins produce partial results continuously is crucial for designing real-time applications.

4

Intermediate

Types of streaming joins in Spark

🤔Before reading on: Which join types do you think Spark supports for streaming joins? Inner only, or also outer and left? Commit your guess.

Concept: Spark supports inner joins, left outer joins, and right outer joins in streaming, each with different behavior on unmatched records.

Inner joins output only matched records from both streams. Left outer joins output all records from the left stream, matching right records if available, else nulls. Right outer joins do the opposite. Full outer joins are not supported in streaming due to complexity.

Result

You know which join types are possible and their effects on streaming data.

Knowing join types helps you choose the right one for your use case and understand output completeness.

5

Intermediate

Handling late and out-of-order data

🤔Before reading on: Do you think streaming joins always wait for all matching data, or can they handle late arrivals? Commit your answer.

Concept: Streaming joins use watermarks and event-time windows to handle late and out-of-order data gracefully.

Watermarks define how late data can arrive before being ignored. Event-time windows group data by time ranges to limit join state size. Spark discards state for windows older than watermark, balancing completeness and resource use.

Result

You understand how Spark manages timing challenges in streaming joins.

Handling late data prevents unbounded memory growth and ensures timely output despite real-world delays.

6

Advanced

State management in streaming joins

🤔Before reading on: Do you think streaming joins keep all past data in memory, or only recent relevant data? Commit your guess.

Concept: Spark manages join state by storing only recent records needed for matching, pruning old state based on watermarks.

Join state holds unmatched records from both streams until matches arrive or state expires. This state is stored in memory or disk and cleaned up periodically. Efficient state management is critical for performance and scalability.

Result

You grasp how Spark balances memory use and correctness in streaming joins.

Understanding state management helps prevent memory leaks and design scalable streaming applications.

7

Expert

Surprises and limitations of streaming joins

🤔Before reading on: Can you guess why full outer joins are not supported in Spark streaming? Commit your answer.

Concept: Certain join types and scenarios are unsupported or tricky due to unbounded state and complexity in streaming joins.

Full outer joins require keeping unmatched records indefinitely, leading to unbounded state. Also, joining two unbounded streams without watermarks can cause infinite state growth. Spark requires at least one stream to have watermarks or be bounded to manage state.

Result

You understand practical limits and design constraints of streaming joins.

Knowing these limits prevents costly mistakes and guides correct streaming join design.

Under the Hood

Streaming joins maintain internal state stores that keep track of unmatched records from each stream keyed by join keys. When a new record arrives on one stream, Spark looks up matching records in the other stream's state. If matches exist, it outputs joined results immediately. State cleanup happens based on watermarks and event-time windows to avoid infinite memory use. The engine uses incremental processing triggered by new data batches, updating state and output continuously.

Why designed this way?

Streaming joins were designed to enable real-time correlation of live data without waiting for all data to arrive, which is impossible for unbounded streams. The use of state and watermarks balances correctness with resource constraints. Alternatives like full outer joins or unbounded state were rejected because they cause memory leaks and unmanageable complexity in continuous processing.

┌─────────────┐       ┌─────────────┐
│ Stream A    │──────▶│ State Store │
└─────────────┘       └─────────────┘
                           │
                           ▼
┌─────────────┐       ┌─────────────┐
│ Stream B    │──────▶│ State Store │
└─────────────┘       └─────────────┘
                           │
                           ▼
                     ┌─────────────┐
                     │ Join Output │
                     └─────────────┘

Myth Busters - 4 Common Misconceptions

Quick: Do streaming joins always produce complete results immediately? Commit yes or no.

Common Belief: Streaming joins produce fully complete joined results as soon as data arrives.

Reality: Streaming joins produce partial results incrementally and may update or drop late matches based on watermarks and state cleanup.

Why it matters: Expecting immediate completeness can lead to wrong assumptions about data freshness and correctness in real-time systems.

Quick: Can you do full outer joins on two unbounded streams in Spark streaming? Commit yes or no.

Common Belief: Full outer joins are supported on any streaming data in Spark.

Reality: Spark does not support full outer joins on two unbounded streams due to unbounded state growth risks.

Why it matters: Trying unsupported joins causes job failures or memory exhaustion in production.

Quick: Do streaming joins keep all past data forever? Commit yes or no.

Common Belief: Streaming joins store all historical data indefinitely to ensure no matches are missed.

Reality: Streaming joins keep only recent data within watermark and window limits, discarding old state to save memory.

Why it matters: Misunderstanding state retention leads to memory leaks or data loss if watermarks are misconfigured.

Quick: Does joining a stream with a static dataset behave the same as joining two streams? Commit yes or no.

Common Belief: Joining a stream with a static dataset is the same as joining two streams.

Reality: Joining a stream with a static dataset is simpler and does not require state management for the static side.

Why it matters: Confusing these can cause inefficient designs and unnecessary complexity.

Expert Zone

1

State cleanup timing depends heavily on watermark accuracy; misconfigured watermarks can cause premature or delayed state eviction.

2

Join keys with high cardinality increase state size dramatically, requiring careful key design and possibly pre-aggregation.

3

The choice between event-time and processing-time joins affects latency and correctness tradeoffs in real-world pipelines.

When NOT to use

Streaming joins are not suitable when full outer joins on unbounded streams are needed or when data arrival is extremely late and unordered beyond watermark limits. Alternatives include batch processing, approximate joins, or using external state stores with custom logic.

Production Patterns

In production, streaming joins are often combined with watermarking and windowing to limit state size. Common patterns include joining user activity streams with reference data, correlating logs from multiple sources, and enriching event streams with static dimension tables.

Connections

Batch joins

Streaming joins build on the same principles as batch joins but adapt them for continuous data.

Understanding batch joins clarifies the core operation streaming joins extend to handle live data.

Event-time processing

Streaming joins rely on event-time semantics to correctly align data from different streams based on when events actually happened.

Mastering event-time helps design joins that handle out-of-order and late data correctly.

Database transaction joins

Streaming joins share conceptual similarities with how databases join tables during transactions but differ in handling unbounded, continuous data.

Comparing streaming joins to database joins highlights challenges of infinite data and state management unique to streaming.

Common Pitfalls

#1 Not setting watermarks causes unbounded state growth.

Wrong approach:

stream1.join(stream2, "key")
.writeStream
.start()

Correct approach:

stream1.withWatermark("eventTime", "10 minutes")
.join(stream2.withWatermark("eventTime", "10 minutes"), "key")
.writeStream
.start()

Root cause:

Without watermarks, Spark cannot know when to remove old state, causing memory to grow indefinitely.

#2 Trying full outer join on two streaming datasets.

Wrong approach:

stream1.join(stream2, "key", "full_outer")
.writeStream
.start()

Correct approach:

Use inner or left/right outer joins with watermarks instead, e.g.
stream1.join(stream2, "key", "inner")
.writeStream
.start()

Root cause:

Full outer joins require keeping unmatched records forever, which is not supported in streaming.

#3 Joining on non-key columns or high-cardinality keys without consideration.

Wrong approach:

stream1.join(stream2, stream1("value") === stream2("value"))
.writeStream
.start()

Correct approach:

Join on well-defined keys with manageable cardinality, e.g.
stream1.join(stream2, "userId")
.writeStream
.start()

Root cause:

Joining on non-key or high-cardinality columns causes large state and poor performance.

Key Takeaways

Streaming joins enable real-time combination of two live data streams by matching records as they arrive.

They differ from batch joins by continuously updating results and managing state for unmatched records.

Watermarks and event-time windows are essential to handle late data and control memory use.

Not all join types are supported in streaming due to unbounded state challenges, especially full outer joins.

Proper state management and join key design are critical for scalable and correct streaming join applications.

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