Solving Spark Data Skew: A Step-by-Step Diagnostic & Implementation Guide
Data skewness is one of the most common production issues in distributed computing, often responsible for increasing infrastructure costs by more than 5x and stalling jobs at the classic 199/200 tasks mark.
Skewness occurs when data is unbalanced across partitions, with a few partitions containing a disproportionately large volume of records compared to the rest. Because a single CPU core processes exactly one partition at a time, skewed data forces a single executor to work infinitely longer than others, leaving your cluster resource utilization highly uneven.
Below is the definitive, step-by-step diagnostic and remediation playbook using PySpark and Salting.
️ Step 1: Diagnose Skewness in the Spark Web UI
Before writing any optimization code, you must first verify that data skewness is indeed the root cause of your bottleneck.
- Open your Spark Web UI (default:
http://localhost:4040). - Go to the Stages tab and click on the active stage that is running slowly or stuck.
- Scroll down to the Summary Metrics for Completed Tasks table.
Under normal execution, task durations and read sizes should be evenly distributed. Under a data skew scenario, you will see a massive disparity between the 75th Percentile and the Max task metrics.
Reading the Diagnostic Metrics:
- Task Duration Distribution: In the graph, 99 tasks completed in under 3 seconds, while a single outlier task (Task 15) is running for 25.2 minutes.
- Duration Metrics:
- Min: 850 ms
- Median (50th Pct): 1.2 s
- 75th Pct: 2.5 s
- Max: 25.1 min ️ This is a 600x skew ratio!
- Shuffle Read Size: The maximum shuffle read is 25.3 GB while the median is 25.3 GB (or the max input size is 250M records vs 100k records for other partitions).
This extreme variance confirms that a single partition contains nearly all the active data, causing CPU Core exhaustion on Executor ID 12.
Step 2: Login to Driver Host & Audit Partition Counts
Once the UI confirms the skew, login to your driver host (or notebook console) to programmatically identify which partitions are overloaded and inspect their row counts.
We use the PySpark built-in function spark_partition_id() to group your DataFrame by partition boundaries:
from pyspark.sql import functions as F
# 1. Audit partition distribution programmatically
audit_df = df.withColumn("partition_id", F.spark_partition_id()) \
.groupBy("partition_id") \
.count() \
.orderBy("partition_id")
# Show the counts for each partition
audit_df.show()
Sample Audit Output:
+------------+------------+
|partition_id|count(rows) |
+------------+------------+
| 0| 999990| <-- ️ Skewed Partition
| 1| 5|
| 2| 5|
+------------+------------+
As the audit reveals, Partition 0 holds 999,990 rows while Partition 1 and 2 hold only 5 rows. During a subsequent join or aggregation, the executor core assigned to Partition 0 will choke while the other executor cores finish instantly and sit idle.
Step 3: Create a Salt Key
To resolve this imbalance, we use Salting.
Why do we need a Salt Key?
By default, Spark uses a hash partitioner to route records during joins. It computes hash(join_key) % num_partitions. If a join key is extremely frequent (e.g., join_key = null or a generic value like 'VIP_USER'), all those records will compute to the exact same hash value and land in the exact same partition.
By adding a random salt key to the join key, we break up the giant hot key into smaller, distinct virtual keys (e.g., VIP_USER_0, VIP_USER_1, VIP_USER_2), forcing Spark to distribute them across separate partitions.
How do we create the Salt Keys?
We define a salt_number (e.g., 3) and apply modifications to both DataFrames being joined:
import sys
from pyspark.sql import functions as F
# Define salt factor based on the number of partitions we want to distribute the skew across
salt_number = 3
# ----------------------------------------------------
# 1. SALT THE LARGE (SKEWED) DATAFRAME
# ----------------------------------------------------
# Add a random integer column between [0, salt number - 1]
large_df_salted = large_df.withColumn(
"salt_key",
F.floor(F.rand() * salt_number)
)
# Create a composite salted join key
large_df_salted = large_df_salted.withColumn(
"salted_join_key",
F.concat(F.col("original_key"), F.lit("_"), F.col("salt_key"))
)
# ----------------------------------------------------
# 2. SALT THE SMALL DATAFRAME (REPLICATION)
# ----------------------------------------------------
# Since the large DF keys are split randomly, the small DF must replicate its rows
# so it can match ANY of the possible random salt values [0, 1, 2].
salt_array = F.array([F.lit(i) for i in range(salt_number)])
small_df_replicated = small_df.withColumn("salt_key_arr", salt_array) \
.withColumn("salt_key", F.explode("salt_key_arr"))
# Create the corresponding composite salted join key
small_df_salted = small_df_replicated.withColumn(
"salted_join_key",
F.concat(F.col("original_key"), F.lit("_"), F.col("salt_key"))
)
️ Step 4: What is the Salt Key Actually Doing?
To understand how salting solves resources starvation, look at how the physical data partitions are reorganized:
BEFORE SALTING (All skewed keys go to one core):
Large DF: [VIP_USER, VIP_USER, VIP_USER, VIP_USER, VIP_USER, VIP_USER] ---> Partition 0 (Core 0 Choking)
Small DF: [VIP_USER] ---> Partition 0
AFTER SALTING (Distributed evenly across multiple cores):
Large DF: [VIP_USER_0, VIP_USER_1, VIP_USER_2, VIP_USER_0, VIP_USER_1, VIP_USER_2]
| | | | | |
v v v v v v
Partition 0 Partition 1 Partition 2 Partition 0 Partition 1 Partition 2
(Core 0) (Core 1) (Core 2) (Core 0) (Core 1) (Core 2)
^ ^ ^ ^ ^ ^
| | | | | |
Small DF: [VIP_USER_0, VIP_USER_1, VIP_USER_2, VIP_USER_0, VIP_USER_1, VIP_USER_2] (Exploded)
By appending a random salt (0, 1, or 2), Spark's hash partitioner routes:
VIP_USER_0records to Partition 0 (processed by Core 0).VIP_USER_1records to Partition 1 (processed by Core 1).VIP_USER_2records to Partition 2 (processed by Core 2).
The intensive join operation is now split across three parallel CPU cores, ensuring uniform hardware utilization.
Step 5: Remove the Salt Key and Output Results
Once the join is successfully computed, the salt keys are no longer needed. We must clean up our schema by dropping the temporary composite keys and salt columns to restore the original table structure.
# 1. Perform the join on the composite salted join key
joined_df = large_df_salted.join(
small_df_salted,
on="salted_join_key",
how="inner"
)
# 2. Drop the temporary salt-keys and composite keys
final_clean_df = joined_df.drop("salted_join_key", "salt_key", "salt_key_arr")
# 3. View audit partition counts post-join to confirm skew resolution
final_clean_df.withColumn("partition_id", F.spark_partition_id()) \
.groupBy("partition_id") \
.count() \
.show()
Sample Final Cleaned Result:
If we fetch a sample of our joined and cleaned DataFrame, it matches your original schema exactly with no trace of the salt variables:
+------------+------------+-----------------+------------------+
|original_key| user_name | action_type | transaction_value|
+------------+------------+-----------------+------------------+
| VIP_USER | Amit Prasad| click_banner | 0.00 |
| VIP_USER | Mukesh Sen | add_to_cart | 89.50 |
| VIP_USER | Amit Prasad| item_checkout| 120.00 |
+------------+------------+-----------------+------------------+
By removing the salt key, the output is perfectly returned to your business format, while the behind-the-scenes processing completed in a fraction of the time!
Follow-Up Questions & Answers
Q1: How do you decide the optimal salt_number?
A: The formula is: salt_number = ceil(skewed_partition_rows / target_partition_size). If the skewed partition has 999,990 rows and you want ~100K rows per partition, use salt_number = 10. Going too high wastes memory by over-replicating the small DataFrame.
Q2: What if BOTH DataFrames are large (1 billion rows each)?
A: Salting still works, but the replication cost of the small table becomes expensive. For two equally large tables:
- Use AQE (Adaptive Query Execution): Enable
spark.sql.adaptive.skewJoin.enabled=truein Spark 3.0+. AQE automatically detects and splits skewed partitions at runtime. - Two-Pass Join: Separate the hot key from the rest, broadcast-join the hot key portion, and do a regular join for the rest.
- Bucketed Tables: Pre-bucket both tables by the join key with the same number of buckets. Spark can do a shuffle-free bucket join.
Q3: Why not just increase spark.sql.shuffle.partitions to fix the skew?
A: Increasing shuffle partitions from 200 to 2000 makes each partition smaller, but it does NOT fix skew. The hot key still hashes to a single partition. If VIP_USER accounts for 60% of data, it will be 60% of whatever partition it lands on — regardless of how many partitions exist.
200 partitions: Partition 47 = 999,990 rows (60%)
2000 partitions: Partition 471 = 999,990 rows (60%) ← Same problem, different partition number!
Salting is necessary because it fundamentally changes the hash by appending a random suffix to the key.
Q4: Can AQE (Adaptive Query Execution) replace salting entirely?
A: For moderate skew (one partition 5-10x larger), AQE is often sufficient. For extreme skew (one partition 100x+ larger), AQE may not split aggressively enough. The safest approach is: enable AQE always, and add salting for known extreme hot keys.
Q5: How do you detect data skew proactively before the job runs?
A: Run a profiling step before the main job:
# Pre-job profiling: detect skew before it crashes the pipeline
key_distribution = df.groupBy("original_key").count()
stats = key_distribution.agg(
F.max("count").alias("max_count"),
F.avg("count").alias("avg_count"),
(F.max("count") / F.avg("count")).alias("skew_ratio")
).collect()[0]
if stats["skew_ratio"] > 10:
print(f"⚠️ SKEW DETECTED: max={stats['max_count']}, avg={stats['avg_count']}, ratio={stats['skew_ratio']:.1f}")
# Automatically enable salting
salt_number = max(3, int(stats["skew_ratio"] / 5))
Sub-Scenarios
Sub-Scenario A: Skew Caused by NULL Keys
Situation: 40% of your records have original_key = NULL. All NULLs hash to the same partition.
Fix: Handle NULLs separately:
null_df = df.filter(F.col("original_key").isNull())
non_null_df = df.filter(F.col("original_key").isNotNull())
# Process non-null keys with normal join
result = non_null_df.join(small_df, on="original_key")
# Handle NULLs based on business logic (drop, assign default, or process separately)
Sub-Scenario B: Skew in a GroupBy Aggregation (Not a Join)
Situation: You're computing SUM(amount) GROUP BY merchant_id, and one merchant has 60% of transactions. The aggregation for that one merchant runs on a single core.
Fix: Use two-stage aggregation with salting:
# Stage 1: Partial aggregation with salt
partial = df.withColumn("salt", F.floor(F.rand() * 10)) \
.groupBy("merchant_id", "salt") \
.agg(F.sum("amount").alias("partial_sum"))
# Stage 2: Final aggregation (removing salt)
final = partial.groupBy("merchant_id") \
.agg(F.sum("partial_sum").alias("total_amount"))
Sub-Scenario C: Skew After a Repartition
Situation: You used df.repartition("merchant_id") to optimize for downstream operations, but now Amazon's 60% of data lands in one partition.
Fix: Use repartition with a salt column or use repartition(200) (round-robin on number of partitions) instead of repartitioning by the skewed key.