Handling skewed joins in Apache Spark - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

Output of a skewed join without optimization

Given two Spark DataFrames df1 and df2 where df2 has a highly skewed key, what will be the output count of the join df1.join(df2, "key")?

Apache Spark

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
data1 = [(1, 'A'), (2, 'B'), (3, 'C'), (4, 'D')]
data2 = [(1, 'X')] * 1000 + [(2, 'Y'), (3, 'Z'), (4, 'W')]
df1 = spark.createDataFrame(data1, ['key', 'val1'])
df2 = spark.createDataFrame(data2, ['key', 'val2'])

joined = df1.join(df2, 'key')
print(joined.count())

A 1003

B 4

C 1000

D 4000

💡 Hint

Think about how many rows match for each key in the join.

The join matches each row in df1 with all matching rows in df2 by key. Key=1 in df2 has 1000 rows, so df1's key=1 row joins with 1000 rows. Other keys have 1 row each. Total is 1000 + 1 + 1 + 1 = 1003.

🧠 Conceptual

intermediate

Reason for skewed join performance issues

Why do skewed joins cause performance problems in distributed data processing frameworks like Spark?

A Because the skewed key causes some tasks to process much more data, leading to imbalance.

B Because skewed joins always cause network failures.

C Because skewed joins reduce the total number of partitions.

D Because skewed joins require more memory on all nodes equally.

💡 Hint

Consider how data is distributed across tasks.

Skewed keys cause some tasks to handle a large portion of data, making those tasks slower and causing overall job delay.

🚀 Application

advanced

Choosing a technique to handle skewed join

You have two large DataFrames with a highly skewed join key. Which technique is best to reduce the impact of skew on join performance?

A Filter out the skewed keys before joining.

B Broadcast the smaller DataFrame to all nodes.

C Increase the number of shuffle partitions without changing data.

D Use salting by adding a random prefix to the skewed key in both DataFrames before joining.

💡 Hint

Think about how to distribute skewed keys evenly.

Salting adds randomness to the skewed key, spreading its data across partitions and reducing task imbalance.

🔧 Debug

advanced

Identify the error in skewed join salting code

What error will this PySpark code raise?

from pyspark.sql.functions import lit, rand, concat

salted_df1 = df1.withColumn('salt', (rand() * 10).cast('int'))
salted_df1 = salted_df1.withColumn('salted_key', concat(df1['key'].cast('string'), salted_df1['salt'].cast('string')))
salted_df2 = df2.withColumn('salt', (rand() * 10).cast('int'))
salted_df2 = salted_df2.withColumn('salted_key', concat(df2['key'].cast('string'), salted_df2['salt'].cast('string')))

joined = salted_df1.join(salted_df2, 'salted_key')

A No error, code runs fine

B SyntaxError due to missing import

C TypeError due to concat of int and string columns without casting

D AnalysisException due to missing join column

💡 Hint

Check the data types used in concat function.

Concat requires string columns. Here, 'key' and 'salt' are integers and need casting to string before concat.

❓ data_output

expert

Result count after salting skewed join

Given df1 with keys [1,2,3] and df2 where key=1 has 100 rows and keys 2,3 have 1 row each, after applying salting with 5 salt values (0 to 4) on key=1 only, what is the count of rows in the salted join?

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import lit, concat, col

spark = SparkSession.builder.getOrCreate()

data1 = [(1, 'A'), (2, 'B'), (3, 'C')]
data2 = [(1, f'X{i}') for i in range(100)] + [(2, 'Y'), (3, 'Z')]

df1 = spark.createDataFrame(data1, ['key', 'val1'])
df2 = spark.createDataFrame(data2, ['key', 'val2'])

# Salting df1 only for key=1
salt_values = [0,1,2,3,4]
salted_df1 = df1.filter(col('key') != 1)
salted_ones = df1.filter(col('key') == 1).crossJoin(spark.createDataFrame([(x,) for x in salt_values], ['salt']))
salted_ones = salted_ones.withColumn('salted_key', concat(col('key').cast('string'), col('salt').cast('string')))
salted_df1 = salted_df1.withColumn('salted_key', col('key').cast('string'))
salted_df1 = salted_df1.union(salted_ones)

# Salting df2 for key=1
salted_df2 = df2.filter(col('key') != 1).withColumn('salted_key', col('key').cast('string'))
salted_ones_df2 = df2.filter(col('key') == 1).crossJoin(spark.createDataFrame([(x,) for x in salt_values], ['salt']))
salted_ones_df2 = salted_ones_df2.withColumn('salted_key', concat(col('key').cast('string'), col('salt').cast('string')))
salted_df2 = salted_df2.union(salted_ones_df2)

joined = salted_df1.join(salted_df2, 'salted_key')
print(joined.count())

A 100

B 500

C 105

D 505

💡 Hint

Calculate how many rows join for salted and non-salted keys.

Key=1 in df1 is duplicated 5 times with salts; df2 key=1 has 100 rows each salted 5 times, so 5*100=500 matches for key=1. Keys 2 and 3 join once each. Total count is 500 + 1 + 1 = 502, closest option is 500.

Practice

1.

What is the main problem caused by skewed joins in Apache Spark?

easy

A. Spark cannot read the input data format

B. Spark runs out of memory due to too many partitions

C. Some keys have many rows, causing slow join performance

D. Data gets duplicated across all nodes

Solution

1. Step 1: Understand skewed join definition

   Skewed joins happen when some keys have a lot more rows than others, causing uneven data distribution.

2. Step 2: Identify the impact on performance

   This uneven distribution causes some tasks to take much longer, slowing down the entire join operation.

3. Final Answer:

   Some keys have many rows, causing slow join performance -> Option C

4. Quick Check:

   Skewed joins = slow join due to uneven keys [OK]

Hint: Skewed join means uneven key distribution slows join [OK]

Common Mistakes:

• Thinking skewed joins cause data duplication
• Confusing skew with memory errors
• Assuming all keys have equal rows

2.

Which of the following is the correct way to add a salt column for handling skewed joins in Spark using PySpark?

from pyspark.sql.functions import rand, floor

# Add salt column
salted_df = df.withColumn('salt', floor(rand() * 10))

easy

A. salted_df = df.withColumn('salt', floor(rand() * 10))

B. salted_df = df.withColumn('salt', rand() + 10)

C. salted_df = df.withColumn('salt', floor(rand()))

D. salted_df = df.withColumn('salt', rand() * 10)

Solution

1. Step 1: Understand salt column creation

   Salt column should be an integer from 0 to 9 to spread keys evenly.

2. Step 2: Check the code correctness

   Using floor(rand() * 10) creates integers 0-9, which is correct for salting.

3. Final Answer:

   salted_df = df.withColumn('salt', floor(rand() * 10)) -> Option A

4. Quick Check:

   floor(rand()*10) creates salt 0-9 [OK]

Hint: Use floor(rand()*N) for integer salt column [OK]

Common Mistakes:

• Using rand() without floor gives float salt
• Adding 10 instead of multiplying
• Not converting to integer with floor

3.

Given the following PySpark code snippet, what will be the output count of joined_df?

df1 = spark.createDataFrame([(1, 'A'), (1, 'B'), (2, 'C')], ['id', 'val1'])
df2 = spark.createDataFrame([(1, 'X'), (1, 'Y'), (3, 'Z')], ['id', 'val2'])

joined_df = df1.join(df2, 'id')
print(joined_df.count())

medium

A. 2

B. 4

C. 3

D. 5

Solution

1. Step 1: Analyze join keys and rows

   df1 has id=1 twice, df2 has id=1 twice, so join on id=1 produces 2*2=4 rows. id=2 and id=3 do not match both sides.

2. Step 2: Calculate total joined rows

   Only id=1 matches, producing 4 rows in joined_df.

3. Final Answer:

   4 -> Option B

4. Quick Check:

   2x2 matches for id=1 = 4 rows [OK]

Hint: Multiply matching rows per key for join count [OK]

Common Mistakes:

• Counting only unique keys, ignoring duplicates
• Assuming join count equals smaller table size
• Ignoring multiple matches per key

4.

Identify the error in the following PySpark code snippet intended to handle skewed joins by salting:

from pyspark.sql.functions import rand, floor

salted_df1 = df1.withColumn('salt', floor(rand() * 5))
salted_df2 = df2.withColumn('salt', floor(rand() * 5))

joined_df = salted_df1.join(salted_df2, ['id', 'salt'])

medium

A. rand() function is not imported correctly

B. floor(rand() * 5) produces values outside 0-4 range

C. Join should be on 'id' only, not on 'salt'

D. Salt columns are added independently, so keys may not match on salt

Solution

1. Step 1: Understand salting logic

   Salt must be added consistently to both dataframes so matching keys have the same salt value.

2. Step 2: Identify error in salt generation

   Here, salt is generated independently with rand(), so salt values won't match for same keys, causing join to fail.

3. Final Answer:

   Salt columns are added independently, so keys may not match on salt -> Option D

4. Quick Check:

   Independent salt columns cause join mismatch [OK]

Hint: Salt must be consistent on both sides for join [OK]

Common Mistakes:

• Adding salt independently without coordination
• Joining only on 'id' when salting is used
• Misunderstanding rand() output range

5.

You have two large datasets with a highly skewed join key. Which approach best handles the skew to improve join performance?

1. Add a random salt column to the skewed dataset and replicate the smaller dataset across all salt values, then join on both key and salt.
2. Disable broadcast joins to force shuffle join.
3. Use map-side join without salting.
4. Increase executor memory without changing join logic.

hard

A. Add random salt to skewed dataset and replicate smaller dataset across salt values, join on key and salt

B. Disable broadcast joins to force shuffle join

C. Use map-side join without salting

D. Increase executor memory without changing join logic

Solution

1. Step 1: Understand skewed join handling

   Salting spreads skewed keys across partitions by adding a random salt column to the skewed dataset and replicating the smaller dataset across all possible salt values.

2. Step 2: Evaluate options

   Disabling broadcast or increasing memory alone won't fix skew. Map-side join without salting won't handle skew properly.

3. Final Answer:

   Add random salt to skewed dataset and replicate smaller dataset across salt values, join on key and salt -> Option A

4. Quick Check:

   Random salt skewed + replicate small fixes skew [OK]

Hint: Random salt skewed + replicate small fixes skew [OK]

Common Mistakes:

• Disabling broadcast join alone fixes skew
• Not replicating the smaller dataset when salting
• Relying only on memory increase