Handling data skew at scale in Apache Spark - Step-by-Step Execution

Concept Flow - Handling data skew at scale

Start: Large Dataset

↓

Detect Skewed Keys

↓

Choose Skew Handling Technique

↓

Salting Keys

↓

Shuffle Data Evenly

↓

Join or Aggregate

↓

Collect Results

↓

End

The flow shows starting with a large dataset, detecting skewed keys, choosing a technique like salting or broadcast join, then processing data evenly to avoid slow tasks.

Execution Sample

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import rand, col, lit, concat
spark = SparkSession.builder.getOrCreate()
df = spark.read.csv('data.csv', header=True)
skewed_key = 'user_id'
salted_df = df.withColumn('salt', (rand() * 10).cast('int'))
salted_df = salted_df.withColumn('salted_key', concat(col(skewed_key), lit('_'), col('salt')))

This code reads data, detects a skewed key, and applies salting by adding a random salt to the key to spread data across partitions.

Execution Table

Step	Action	Variable/Column	Value/Result	Effect
1	Read CSV file	df.count()	1000000	Loaded 1 million rows
2	Identify skewed key	skewed_key	'user_id'	User ID causes skew
3	Add salt column	salt	Random int 0-9	Salt spreads keys
4	Create salted key	salted_key	user_id + '_' + salt	Keys now unique per salt
5	Shuffle data on salted_key	Partitions	Balanced	Avoids skew in shuffle
6	Join or aggregate on salted_key	Result	Correct and balanced	Faster execution
7	Remove salt after join	Final keys	Original user_id	Restore original keys
8	End	-	-	Skew handled successfully

💡 Salted keys balance data, preventing slow tasks caused by skewed user_id distribution.

Variable Tracker

Variable/Column	Start	After Step 3	After Step 4	After Step 7	Final
df.count()	N/A	N/A	N/A	N/A	1000000
skewed_key	N/A	'user_id'	'user_id'	'user_id'	'user_id'
salt	N/A	Random int 0-9	Random int 0-9	Dropped	Dropped
salted_key	N/A	N/A	user_id + '_' + salt	Dropped	Dropped

Key Moments - 3 Insights

Why do we add a 'salt' column to the skewed key?

Adding a salt creates multiple versions of the skewed key, spreading data across partitions evenly. This avoids one partition having too much data, as shown in execution_table step 3 and 4.

How do we restore original keys after processing with salted keys?

After the join or aggregation, we remove the salt column to get back the original keys, as shown in execution_table step 7. This ensures final results use the correct keys.

When is broadcast join preferred over salting?

Broadcast join is used when one table is small enough to fit in memory, avoiding shuffle. Salting is for large skewed keys in big tables. This choice is part of the decision in concept_flow.

Visual Quiz - 3 Questions

Look at the execution_table at step 4, what does the 'salted_key' column contain?

A Original user_id only

B Concatenation of user_id and salt

C Random salt only

D Null values

💡 Hint

Refer to execution_table row 4 under 'Value/Result' column.

At which step does the data shuffle become balanced to avoid skew?

A Step 7

B Step 2

C Step 5

D Step 1

💡 Hint

Check execution_table row 5 under 'Effect' column.

If we skip adding the salt column, what likely happens to the shuffle partitions?

A Partitions become skewed with some very large

B Partitions become balanced

C Partitions are empty

D Partitions double in number

💡 Hint

Refer to key_moments question 1 about the purpose of salting.

Concept Snapshot

Handling data skew at scale in Spark:
- Detect skewed keys causing uneven data distribution.
- Use salting: add random salt to keys to spread data.
- Or use broadcast join if one table is small.
- Shuffle data on salted keys for balance.
- Remove salt after processing to restore keys.
- This improves performance by avoiding slow tasks.

Full Transcript

This visual execution shows how to handle data skew at scale in Apache Spark. We start by loading a large dataset and identifying a skewed key, such as user_id. To fix skew, we add a salt column with random integers and create a salted key by combining user_id and salt. This spreads data evenly across partitions during shuffle, preventing slow tasks caused by skew. After join or aggregation, we remove the salt to restore original keys. Alternatively, if one table is small, we use broadcast join to avoid shuffle. The execution table traces each step, showing variable changes and effects. Key moments clarify why salting is needed and how keys are restored. The quiz tests understanding of salted keys and shuffle balance. This method helps Spark jobs run faster and more reliably on skewed data.

Practice

1. What is the main problem caused by data skew in Apache Spark jobs?

easy

A. Some tasks become much larger and slower than others

B. All tasks run equally fast

C. Data is lost during processing

D. Spark automatically fixes skew without user action

Solution

1. Step 1: Understand data skew impact

   Data skew means some keys have many more records, causing uneven task sizes.

2. Step 2: Identify effect on tasks

   Tasks processing large keys take longer, slowing the whole job.

3. Final Answer:

   Some tasks become much larger and slower than others -> Option A

4. Quick Check:

   Data skew = uneven task sizes [OK]

Hint: Data skew means uneven task sizes slowing jobs [OK]

Common Mistakes:

• Thinking all tasks run equally fast
• Believing Spark fixes skew automatically
• Confusing data loss with skew

2. Which of the following is the correct way to add a salt key to a DataFrame column named user_id in PySpark?

easy

A. df.withColumn('salted_key', concat(col('user_id'), lit('_'), rand()))

B. df.withColumn('salted_key', col('user_id') + rand())

C. df.withColumn('salted_key', col('user_id').cast('int'))

D. df.withColumn('salted_key', col('user_id').substr(0, 3))

Solution

1. Step 1: Understand salting technique

   Salting adds a random value to the key to spread data across partitions.

2. Step 2: Check correct syntax for salting

   Concatenating user_id with a random number as string is correct.

3. Final Answer:

   df.withColumn('salted_key', concat(col('user_id'), lit('_'), rand())) -> Option A

4. Quick Check:

   Salting = concat key + random [OK]

Hint: Use concat with random value to salt keys [OK]

Common Mistakes:

• Adding random float directly to string key
• Casting key without salting
• Using substring instead of salting

3. Given the following PySpark code snippet to reduce skew by salting, what will be the output count of salted_df if df has 1000 rows?

from pyspark.sql.functions import col, concat, lit, rand
salted_df = df.withColumn('salted_key', concat(col('key'), lit('_'), (rand() * 10).cast('int')))
count = salted_df.count()

medium

A. 0

B. 1000

C. 10

D. 100

Solution

1. Step 1: Understand salting effect on row count

   Salting adds a new column but does not change the number of rows.

2. Step 2: Count rows after salting

   The count remains the same as original DataFrame, 1000 rows.

3. Final Answer:

   1000 -> Option B

4. Quick Check:

   Salting adds column, count unchanged [OK]

Hint: Salting adds column, row count stays same [OK]

Common Mistakes:

• Thinking salting duplicates rows
• Confusing salt range with row count
• Assuming count changes after withColumn

4. You have a join in Spark that is very slow due to data skew on the join key. Which of the following code snippets correctly applies salting to both DataFrames to fix the skew?

df1_salted = df1.withColumn('salt', (rand() * 5).cast('int'))
df2_salted = df2.withColumn('salt', (rand() * 5).cast('int'))
joined = df1_salted.join(df2_salted, ['key', 'salt'])

medium

A. Incorrect: salting with fixed value instead of random

B. Incorrect: salting only one DataFrame

C. Incorrect: joining without salt column

D. Correct: salting both sides and joining on key and salt

Solution

1. Step 1: Understand salting for joins

   To fix skew, salt both DataFrames on join key with same salt range.

2. Step 2: Verify join condition

   Joining on both original key and salt column ensures balanced partitions.

3. Final Answer:

   Correct: salting both sides and joining on key and salt -> Option D

4. Quick Check:

   Salt both sides + join on key & salt [OK]

Hint: Salt both DataFrames and join on key plus salt [OK]

Common Mistakes:

• Salting only one DataFrame
• Joining without salt column
• Using fixed salt value instead of random

5. You have a Spark job with a very large skewed key causing one partition to be huge. You want to reduce skew by salting and also control shuffle partitions. Which approach below best balances load and parallelism?

hard

A. Add salt column but keep shuffle partitions very low

B. Do not add salt, just increase shuffle partitions

C. Add a salt column with random values, join on salted keys, and increase shuffle partitions

D. Add salt column with fixed value and reduce shuffle partitions

Solution

1. Step 1: Apply salting to spread skewed key

   Adding a salt column with random values breaks large groups into smaller ones.

2. Step 2: Adjust shuffle partitions to increase parallelism

   Increasing shuffle partitions allows more tasks to run in parallel, reducing bottlenecks.

3. Step 3: Combine salting and partition tuning

   Both salting and increasing partitions together balance load and improve job speed.

4. Final Answer:

   Add a salt column with random values, join on salted keys, and increase shuffle partitions -> Option C

5. Quick Check:

   Salt + more partitions = balanced load [OK]

Hint: Salt keys and increase shuffle partitions for best balance [OK]

Common Mistakes:

• Keeping shuffle partitions too low after salting
• Not salting and only increasing partitions
• Using fixed salt value causing no skew fix