Data Skew & Salting

Data Skew is one of the most common and difficult performance issues in distributed computing. It occurs when a dataset's keys are unevenly distributed, causing a few partitions to hold a massive percentage of the data while other partitions are nearly empty.

Without Salting:
Key: "heavy_key"  Hash  All land on Partition 1 (5 GB)  Task 1 runs for 4 hours!
Key: "small_key"  Hash  Lands on Partition 2 (5 MB)   Task 2 runs in 2 seconds!

The Impact of Data Skew

• The "99% Stuck" Problem: Your Spark job runs extremely fast at first, but then hangs at 99% progress bar for hours. A single executor thread is struggling to sort/process a massive skewed partition.
• Out of Memory (OOM) Crashes: Executors running skewed partitions exceed their JVM heap memory limits and crash, causing job failures.

Resolving Skew via Salting

Salting is an optimization technique that breaks up a massive, skewed key by appending a random suffix (the "salt") to it. This forces Spark to distribute the records sharing that key across multiple distinct partitions to process them in parallel.

The Salting Algorithm:

1. Skewed Table: Add a random integer suffix between 0 and N-1 (e.g., user_123 becomes user_123_0 or user_123_1).
2. Dimension Table: Replicate (explode) every row N times, appending all possible suffixes (0 to N-1) to match the salted keys.
3. The Join: Join on the salted keys. Spark hashes these salted keys, spreading the skewed rows across N parallel executor tasks!

PySpark Code Example: Implementing Salting

Here is a complete, detailed implementation of the Salting optimization technique in PySpark:

import random
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

# 1. Setup Spark
spark = SparkSession.builder \
    .appName("Salting Joins") \
    .master("local[*]") \
    .getOrCreate()

# We define N = 3 salts to distribute the skew across 3 parallel tasks
num_salts = 3

# 2. Large, Skewed Transactions DataFrame
# The product key 'Laptop' is highly skewed (has 1,000,000 sales)
skewed_data = [("Laptop", 1000.0) for _ in range(1000000)] + [("Mouse", 20.0), ("Keyboard", 50.0)]
large_df = spark.createDataFrame(skewed_data, ["product_key", "amount"])

# Apply Salting: Add a random suffix ' 0', ' 1', or ' 2' to the keys
salted_large_df = large_df.withColumn(
    "salted_key",
    F.concat(F.col("product_key"), F.lit("_"), F.rand(seed=42).multiply(num_salts).cast("int"))
)

# 3. Small Dimension DataFrame (Products Metadata)
dim_data = [("Laptop", "Tech Department"), ("Mouse", "Office Accessories"), ("Keyboard", "Office Accessories")]
dim_df = spark.createDataFrame(dim_data, ["product_key", "department"])

# Prepare Dimension Table: Explode every row N times with all possible suffixes
# We first create an array containing [0, 1, 2] and explode it
salt_array = F.array([F.lit(i) for i in range(num_salts)])
salted_dim_df = dim_df.withColumn("salt", F.explode(salt_array)) \
                      .withColumn("salted_key", F.concat(F.col("product_key"), F.lit("_"), F.col("salt")))

# 4. Perform the Join on the Salted Keys
# The 1,000,000 'Laptop' records are now split across 3 parallel join tasks!
optimized_joined_df = salted_large_df.join(
    salted_dim_df,
    "salted_key",
    "inner"
).drop("salted_key", "salt")

optimized_joined_df.show(5)