Handling skewed joins in Apache Spark - Step-by-Step Execution

Concept Flow - Handling skewed joins

Start Join Operation

↓

Detect Skewed Keys?

No→Perform Normal Join

Yes↓

Apply Skew Join Handling

↓

Split Skewed Keys

↓

Join Skewed Keys Separately

↓

Combine Results

↓

Output Result

The join operation first checks for skewed keys. If none, it performs a normal join. If skewed keys exist, it splits and joins them separately, then combines results.

Execution Sample

Apache Spark

from pyspark.sql.functions import broadcast

skewed_keys = ['key1']

# Split skewed keys
skewed_df = df1.filter(df1.key.isin(skewed_keys))
rest_df = df1.filter(~df1.key.isin(skewed_keys))

# Join skewed keys with broadcast
join_skewed = skewed_df.join(broadcast(df2), 'key')

# Join rest normally
join_rest = rest_df.join(df2, 'key')

# Combine results
final_join = join_skewed.union(join_rest)

This code splits the data on skewed keys, joins skewed keys with broadcast join, joins the rest normally, then combines results.

Execution Table

Step	Action	DataFrame State	Join Type	Output Rows
1	Identify skewed keys	skewed_keys = ['key1']	N/A	N/A
2	Filter skewed keys from df1	skewed_df contains rows with key1	N/A	N/A
3	Filter rest keys from df1	rest_df contains rows without key1	N/A	N/A
4	Join skewed_df with df2 using broadcast	Joining skewed_df and df2 on key1	Broadcast Join	1000 rows (skewed key)
5	Join rest_df with df2 normally	Joining rest_df and df2 on other keys	Shuffle Join	9000 rows (non-skewed keys)
6	Combine join_skewed and join_rest	Union of both join results	N/A	10000 rows total
7	End of join operation	Final joined DataFrame ready	N/A	10000 rows total

💡 All rows joined; skewed keys handled separately to avoid performance issues.

Variable Tracker

Variable	Start	After Step 2	After Step 3	After Step 4	After Step 5	Final
skewed_keys	[]	['key1']	['key1']	['key1']	['key1']	['key1']
skewed_df	empty	rows with key1	rows with key1	rows with key1	rows with key1	rows with key1
rest_df	empty	empty	rows without key1	rows without key1	rows without key1	rows without key1
join_skewed	empty	empty	empty	joined rows with key1	joined rows with key1	joined rows with key1
join_rest	empty	empty	empty	empty	joined rows without key1	joined rows without key1
final_join	empty	empty	empty	empty	empty	union of skewed and rest joins

Key Moments - 3 Insights

Why do we split the data into skewed and non-skewed parts before joining?

Splitting allows us to handle skewed keys separately with broadcast join, avoiding heavy shuffles and improving performance, as shown in steps 2-5 of the execution table.

What happens if we do not handle skewed keys separately?

The join will cause data skew, leading to some tasks taking much longer and slowing down the whole job, because the large skewed key causes uneven data distribution during shuffle.

Why do we use broadcast join for skewed keys?

Broadcast join sends the smaller DataFrame to all nodes, avoiding shuffle for skewed keys, which reduces data movement and speeds up the join, as seen in step 4.

Visual Quiz - 3 Questions

Look at the execution table, what type of join is used for the skewed keys?

A Broadcast Join

B Shuffle Join

C Sort Merge Join

D Cartesian Join

💡 Hint

Check step 4 in the execution table where skewed keys are joined.

At which step do we combine the results of skewed and non-skewed joins?

A Step 3

B Step 6

C Step 5

D Step 7

💡 Hint

Look for the step mentioning union of join results.

If we did not filter out skewed keys before joining, what would likely happen?

A The join would be faster

B The join would produce fewer rows

C The join would cause data skew and slow down

D The join would fail with error

💡 Hint

Refer to the key moment about consequences of not handling skewed keys.

Concept Snapshot

Handling skewed joins in Spark:
- Detect skewed keys causing uneven data distribution
- Split data into skewed and non-skewed parts
- Join skewed keys using broadcast join
- Join rest normally
- Combine results to get final joined DataFrame
This avoids slow shuffle caused by skew.

Full Transcript

In Spark, skewed joins happen when some keys have many more rows than others, causing slow joins. To handle this, we first detect skewed keys. Then we split the data into two parts: one with skewed keys and one without. We join the skewed keys separately using a broadcast join, which avoids heavy shuffles. The rest of the data is joined normally. Finally, we combine both join results. This method improves performance by balancing the workload across the cluster.