Handling Data Skew in Apache Spark Using Salting: A Practical Approach to OOM Errors
Learn how data skew in Apache Spark can cause uneven partitions, shuffle overhead, slow tasks, and potential OOM issues, and how salting can help distribute skewed data more efficiently using PySpark.
Handling Data Skew in Apache Spark Using Salting: A Practical Approach to OOM Errors
Apache Spark is widely used for processing large datasets because it can distribute workloads across multiple executors and partitions. However, distributed processing can face performance and memory problems when data is not evenly distributed.
One common problem is data skew.
Data skew can cause some Spark partitions to contain significantly more data than others. As a result, one or more tasks may take much longer than the rest and consume a large amount of executor memory. In extreme cases, this can contribute to Out-of-Memory (OOM) errors.
One technique that can help address this problem is salting.
What is Data Skew?
Data skew occurs when data is distributed unevenly across partitions.
For example, imagine a dataset containing customer transactions:
| customer_id | Number of records |
|---|---|
| 1001 | 10,000,000 |
| 1002 | 500 |
| 1003 | 300 |
| 1004 | 200 |
Here, customer
1001 represents a highly frequent or skewed key.When Spark performs operations such as joins, aggregations, or
groupBy() using customer_id, a large amount of data associated with 1001 can end up being processed by a small number of partitions.This creates an imbalance in the workload.
Why Can Data Skew Cause OOM?
Spark divides data into partitions and processes those partitions in parallel.
Ideally:
1
2
3
4
Partition 1 → 1 GB
Partition 2 → 1 GB
Partition 3 → 1 GB
Partition 4 → 1 GB
The workload is relatively balanced.
With data skew, we might have:
1
2
3
4
Partition 1 → 100 MB
Partition 2 → 120 MB
Partition 3 → 150 MB
Partition 4 → 4 GB
The fourth partition has much more data than the others.
The task processing this partition may require significantly more memory. During operations such as joins and aggregations, Spark may also need additional memory for shuffle data, sorting, hash tables, and intermediate results.
This can lead to:
1
2
3
4
5
6
7
8
9
10
11
Data Skew
↓
Uneven Partitions
↓
Large Shuffle Partition
↓
High Memory Consumption
↓
Slow Task / Executor Memory Pressure
↓
Potential OOM
It is important to note that data skew is not the only cause of OOM errors. Large datasets, inappropriate partition sizes, expensive transformations, caching, and insufficient executor memory can also cause OOM.
What is Salting?
Salting is a technique used to distribute records associated with a heavily skewed key across multiple partitions.
Instead of using only:
1
customer_id
as the key, we introduce an additional column called a salt column.
For example:
1
2
3
4
5
6
7
customer_id | salt
------------|-----
1001 | 0
1001 | 1
1001 | 2
1001 | 3
1001 | 4
Now the heavily skewed customer
1001 can be distributed across multiple salted keys.Instead of processing everything using:
1
1001
we can process:
1
2
3
4
5
1001_0
1001_1
1001_2
1001_3
1001_4
This can help distribute the workload more evenly.
Simple PySpark Example
Let's create a sample dataset with a highly frequent customer.
1
2
3
4
5
6
7
8
9
10
11
12
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, floor, rand
spark = SparkSession.builder \
.appName("DataSkewExample") \
.getOrCreate()
df = spark.read.csv(
"sales.csv",
header=True,
inferSchema=True
)
Suppose
customer_id = 1001 appears millions of times.We can create a salt column:
1
2
3
4
5
6
salt_count = 10
df_salted = df.withColumn(
"salt",
floor(rand() * salt_count)
)
Now the data may look like:
1
2
3
4
5
6
7
customer_id | amount | salt
------------|--------|-----
1001 | 500 | 0
1001 | 200 | 3
1001 | 150 | 7
1001 | 300 | 2
1001 | 450 | 9
The salt value is randomly distributed between
0 and 9.Salting During a Join
Salting becomes particularly useful when dealing with a skewed join.
Suppose we have:
1
2
3
4
5
Large Sales Table
|
| customer_id
↓
Customer Table
If one customer appears millions of times in the sales table, the join can become skewed.
The basic join is:
1
2
3
4
5
result = sales.join(
customers,
sales.customer_id == customers.customer_id,
"inner"
)
With salting, we can create a salt column on the large dataset and replicate the smaller lookup dataset across the required salt values.
Conceptually:
1
2
3
4
5
6
7
8
9
Large Table
customer_id | salt
------------|-----
1001 | 0
1001 | 1
1001 | 2
1001 | 3
...
And the smaller table can be expanded:
1
2
3
4
5
6
7
customer_id | salt
------------|-----
1001 | 0
1001 | 1
1001 | 2
1001 | 3
...
The join can then use:
1
2
3
4
5
result = large_df.join(
small_df,
["customer_id", "salt"],
"inner"
)
The important idea is that the original skewed key is combined with the salt value, creating multiple keys and allowing Spark to distribute the workload more evenly.
How to Identify Data Skew
Before applying salting, we should first investigate whether skew actually exists.
One simple approach is to count records by the join key:
1
2
3
4
df.groupBy("customer_id") \
.count() \
.orderBy(col("count").desc()) \
.show()
If the result looks like:
1
2
3
4
5
6
customer_id | count
------------|----------
1001 | 10000000
1002 | 500
1003 | 300
1004 | 200
then
1001 is a potential skewed key.Spark UI can also be useful for identifying skew. When examining a job or stage, look for tasks whose execution time or input/shuffle size is significantly larger than the other tasks.
For example:
1
2
3
4
Task 1 → 2 seconds
Task 2 → 3 seconds
Task 3 → 2 seconds
Task 4 → 45 seconds
A large difference between tasks can be an indication that the workload is not evenly distributed.
Salting Is Not Always the Best Solution
Although salting can be useful, it should not automatically be the first solution.
Depending on the situation, other approaches may be better.
1. Broadcast Join
If one side of the join is small enough, a broadcast join can avoid a large shuffle.
1
2
3
4
5
6
from pyspark.sql.functions import broadcast
result = large_df.join(
broadcast(small_df),
"customer_id"
)
2. Adaptive Query Execution
Modern Spark versions provide Adaptive Query Execution (AQE), which can dynamically optimize query execution.
AQE can help with certain types of skewed joins.
3. Repartitioning
Repartitioning can sometimes improve data distribution:
1
df = df.repartition("customer_id")
However, simply repartitioning does not necessarily solve a highly skewed key. If the same key remains dominant, the partition can still be large.
4. Filter Unnecessary Data
Reducing the amount of data before a join or aggregation can significantly reduce memory and shuffle requirements.
For example:
1
df = df.filter(col("status") == "ACTIVE")
It is generally better to filter early when the filter is valid for the business logic.
When Should You Consider Salting?
Salting can be considered when:
- A few keys contain a very large percentage of the data.
- A join is suffering from severe data skew.
- One or a few tasks are significantly slower than others.
- Shuffle partitions are highly unbalanced.
- Broadcast join is not practical.
- Other optimization techniques do not sufficiently address the skew.
Key Takeaways
Data skew is an important performance challenge in distributed data processing.
The basic relationship is:
1
2
3
4
5
6
7
8
9
Skewed Data
↓
Uneven Partition Distribution
↓
Large / Slow Tasks
↓
High Shuffle and Memory Pressure
↓
Potential Performance Problems or OOM
Salting introduces an additional value to a skewed key so that the workload can be distributed across multiple partitions.
However, salting should be used based on actual evidence of skew rather than applied blindly.
Before implementing salting, investigate the data distribution, examine Spark UI metrics, understand the join strategy, and consider alternatives such as broadcast joins, AQE, filtering, and appropriate partitioning.
Understanding data skew, shuffle, partitions, joins, and memory usage is an important part of becoming effective with Apache Spark and PySpark for large-scale data engineering workloads.
Enjoyed reading this content? Let the author know!
Your likes, comments, shares, and saves help creators reach more builders.
Loading recommendations
Loading article