Eric Campbell 5 min read Engineering

Apache Spark - Salting Technique for Skewed Data

Data skew is the silent killer of Spark job performance. It occurs when data isn't uniformly distributed across partition keys, causing some partitions to contain orders of magnitude more records...

Key Insights

  • Data skew causes a single Spark task to process disproportionately more data than others, leading to job stragglers that can extend execution time by 10-100x while other executors sit idle.
  • Salting artificially distributes skewed keys across multiple partitions by appending random prefixes, trading increased shuffle volume for dramatically improved parallelism.
  • Selective salting—applying the technique only to known hot keys—minimizes overhead while still eliminating the bottleneck, and can be combined with Spark 3.x Adaptive Query Execution for optimal results.

Understanding Data Skew in Spark

Data skew is the silent killer of Spark job performance. It occurs when data isn’t uniformly distributed across partition keys, causing some partitions to contain orders of magnitude more records than others. When Spark shuffles data for operations like joins or aggregations, one unlucky executor gets stuck processing a massive partition while the rest finish quickly and wait.

The impact is severe. A join that should complete in 5 minutes takes 2 hours because a single task is processing 90% of the data. You’ll see OOM errors when that one partition exceeds executor memory. Your cluster utilization plummets as 99 executors sit idle waiting for the straggler.

Real-world scenarios where skew appears:

  • E-commerce transactions: A few popular products account for 80% of orders
  • Null or default values: Missing data concentrated in a single key
  • Geographic data: Major cities dominate location-based datasets
  • Time-series data: Viral events create massive spikes on specific dates
  • Customer analytics: Power users generate disproportionate activity

Identifying Data Skew

Before applying fixes, confirm skew is your actual problem. The Spark UI tells the story clearly.

Symptoms in the Spark UI:

  • Task duration variance: One task takes 30 minutes while others complete in seconds
  • Shuffle read size imbalance: One task reads 50GB while others read 50MB
  • Executor timeline showing a single long-running task

Diagnostic code to find skewed keys:

from pyspark.sql import functions as F

# Analyze key distribution in your dataset
key_distribution = (
    transactions_df
    .groupBy("product_id")
    .agg(
        F.count("*").alias("record_count"),
        F.sum("amount").alias("total_amount")
    )
    .orderBy(F.desc("record_count"))
)

# Show top skewed keys
key_distribution.show(20)

# Calculate skew metrics
stats = key_distribution.select(
    F.mean("record_count").alias("mean"),
    F.stddev("record_count").alias("stddev"),
    F.max("record_count").alias("max"),
    F.min("record_count").alias("min")
).collect()[0]

print(f"Mean records per key: {stats['mean']:.0f}")
print(f"Max records for single key: {stats['max']}")
print(f"Skew ratio (max/mean): {stats['max']/stats['mean']:.1f}x")

A skew ratio above 10x typically indicates a problem worth addressing. Above 100x, you’re almost certainly seeing significant performance degradation.

How Salting Works

Salting is conceptually simple: you artificially expand skewed keys by appending random suffixes, distributing what was one massive partition across many smaller ones.

Consider a join between transactions and products where product_id = "POPULAR_ITEM" has 10 million records while most products have under 1,000. Without salting, one task handles all 10 million records.

With a salt factor of 10:

  • The skewed key becomes POPULAR_ITEM_0 through POPULAR_ITEM_9
  • Each salted key gets roughly 1 million records
  • 10 tasks now process the work in parallel

The trade-off is real: you’re increasing data volume. The smaller table (products) must be replicated for each salt value. If your salt factor is 10, you’re creating 10 copies of product records. This increases shuffle volume but dramatically improves parallelism.

Before salting:

Partition 0: [ITEM_A: 100 records]
Partition 1: [ITEM_B: 150 records]
Partition 2: [POPULAR_ITEM: 10,000,000 records]  <- Straggler
Partition 3: [ITEM_C: 80 records]

After salting (factor=4):

Partition 0: [ITEM_A: 100, POPULAR_ITEM_0: 2,500,000]
Partition 1: [ITEM_B: 150, POPULAR_ITEM_1: 2,500,000]
Partition 2: [ITEM_C: 80, POPULAR_ITEM_2: 2,500,000]
Partition 3: [POPULAR_ITEM_3: 2,500,000]

Implementing Basic Salting

Here’s a complete implementation for salting a skewed join:

from pyspark.sql import functions as F
from pyspark.sql.types import IntegerType

SALT_FACTOR = 10

# Large table with skewed keys (e.g., transactions)
transactions_df = spark.table("transactions")

# Smaller dimension table (e.g., products)
products_df = spark.table("products")

# Step 1: Add random salt to the large (skewed) table
salted_transactions = transactions_df.withColumn(
    "salt",
    (F.rand() * SALT_FACTOR).cast(IntegerType())
).withColumn(
    "salted_product_id",
    F.concat(F.col("product_id"), F.lit("_"), F.col("salt"))
)

# Step 2: Explode the smaller table to match all possible salts
salt_values = spark.range(SALT_FACTOR).withColumnRenamed("id", "salt")

exploded_products = (
    products_df
    .crossJoin(salt_values)
    .withColumn(
        "salted_product_id",
        F.concat(F.col("product_id"), F.lit("_"), F.col("salt"))
    )
)

# Step 3: Join on salted keys
result = salted_transactions.join(
    exploded_products,
    on="salted_product_id",
    how="inner"
).drop("salt", "salted_product_id")

Scala equivalent:

val SALT_FACTOR = 10

val saltedTransactions = transactions
  .withColumn("salt", (rand() * SALT_FACTOR).cast(IntegerType))
  .withColumn("salted_product_id", 
    concat(col("product_id"), lit("_"), col("salt")))

val saltValues = spark.range(SALT_FACTOR).withColumnRenamed("id", "salt")

val explodedProducts = products
  .crossJoin(saltValues)
  .withColumn("salted_product_id",
    concat(col("product_id"), lit("_"), col("salt")))

val result = saltedTransactions
  .join(explodedProducts, Seq("salted_product_id"), "inner")
  .drop("salt", "salted_product_id")

Selective Salting for Known Hot Keys

Full salting is wasteful when only a few keys cause problems. Selective salting applies the technique only to identified hot keys, minimizing overhead.

from pyspark.sql import functions as F

# Known problematic keys (identified from analysis)
HOT_KEYS = ["POPULAR_ITEM_1", "POPULAR_ITEM_2", "VIRAL_PRODUCT"]
SALT_FACTOR = 20

# Create broadcast set for efficient lookup
hot_keys_set = set(HOT_KEYS)
is_hot_key = F.col("product_id").isin(HOT_KEYS)

# Salt only hot keys in large table
salted_transactions = transactions_df.withColumn(
    "salt",
    F.when(is_hot_key, (F.rand() * SALT_FACTOR).cast(IntegerType()))
     .otherwise(F.lit(0))
).withColumn(
    "salted_product_id",
    F.when(is_hot_key, 
           F.concat(F.col("product_id"), F.lit("_"), F.col("salt")))
     .otherwise(F.col("product_id"))
)

# Explode only hot keys in small table
salt_range = list(range(SALT_FACTOR))

exploded_products = products_df.withColumn(
    "salt_array",
    F.when(F.col("product_id").isin(HOT_KEYS), 
           F.array([F.lit(i) for i in salt_range]))
     .otherwise(F.array(F.lit(0)))
).withColumn(
    "salt", 
    F.explode("salt_array")
).withColumn(
    "salted_product_id",
    F.when(F.col("product_id").isin(HOT_KEYS),
           F.concat(F.col("product_id"), F.lit("_"), F.col("salt")))
     .otherwise(F.col("product_id"))
).drop("salt_array")

# Join on salted keys
result = salted_transactions.join(
    exploded_products,
    on="salted_product_id",
    how="inner"
).drop("salt", "salted_product_id")

Hybrid approach for very small hot-key subsets:

When hot keys have few corresponding records in the dimension table, broadcast those separately:

# Separate hot and cold data
hot_products = products_df.filter(F.col("product_id").isin(HOT_KEYS))
cold_products = products_df.filter(~F.col("product_id").isin(HOT_KEYS))

hot_transactions = transactions_df.filter(F.col("product_id").isin(HOT_KEYS))
cold_transactions = transactions_df.filter(~F.col("product_id").isin(HOT_KEYS))

# Broadcast join for hot keys (small dimension data)
hot_result = hot_transactions.join(
    F.broadcast(hot_products),
    on="product_id",
    how="inner"
)

# Regular join for cold keys (no skew)
cold_result = cold_transactions.join(
    cold_products,
    on="product_id", 
    how="inner"
)

# Union results
result = hot_result.unionByName(cold_result)

Performance Tuning and Best Practices

Choosing the optimal salt factor:

  • Start with sqrt(skew_ratio) as a baseline
  • Too low: Still leaves partitions too large
  • Too high: Excessive shuffle from exploded dimension table
  • Test with values like 10, 20, 50 and measure actual task distribution

Combining with Adaptive Query Execution (Spark 3.x):

AQE can automatically handle some skew scenarios, but salting remains valuable:

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB")

AQE’s automatic skew handling works well for moderate skew but may not fully resolve extreme cases. Use both: let AQE handle minor skew while salting addresses severe hot keys.

When NOT to use salting:

  • Small dimension tables: Use broadcast joins instead
  • Uniform skew: If all keys are roughly equally large, repartition with more partitions
  • Aggregations without joins: Consider partial aggregation or two-phase aggregation
  • Moderate skew: AQE in Spark 3.x often handles this automatically

Conclusion

Salting is a surgical fix for data skew in Spark joins. It trades controlled data amplification for parallelism gains that can reduce job runtime from hours to minutes.

Implementation checklist:

  1. Confirm skew is your bottleneck using Spark UI and key distribution analysis
  2. Identify specific hot keys causing the problem
  3. Choose selective salting for known hot keys, full salting only when necessary
  4. Start with a salt factor around sqrt(skew_ratio) and tune based on results
  5. Enable AQE as a complementary optimization
  6. Monitor post-implementation to verify improvement and adjust salt factor

Don’t apply salting blindly to all joins. Profile first, salt the specific joins with confirmed skew, and measure the results. When applied correctly, this technique transforms unusable Spark jobs into production-ready pipelines.

Liked this? There's more.

Every week: one practical technique, explained simply, with code you can use immediately.