Workload Scenarios:

Shuffle Stage & Cluster Inputs

5 GB500 GB1 TB2 TB

Shuffle Partitions & Salting Sizing

spark.sql.shuffle.partitions
1200
Targeting ~128MB per task
Recommended Salt Factor
50 buckets
Distributes skewed join keys
Straggler Task Hazard (High)Severe data skew detected: 35% of data concentrates on single keys. One executor task will process 52.5 GB and straggle, causing stage timeouts.

Optimized PySpark Shuffle & Salting Code

# 1. Enable Adaptive Query Execution (AQE) in Spark 3.x
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "134217728")
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", "268435456")
spark.conf.set("spark.sql.shuffle.partitions", "1200")

# 2. Salting Transformation to eliminate stragglers on 'user_id'
from pyspark.sql import functions as F

SALT_FACTOR = 50

# Salt the skewed Large DataFrame (LHS)
df_left_salted = df_large.withColumn(
    "salted_key",
    F.concat(F.col("user_id"), F.lit("_"), F.floor(F.rand() * SALT_FACTOR))
)

# Replicate the Dimension/Right DataFrame (RHS) across all salt buckets
df_right_replicated = df_small.withColumn(
    "salt_array",
    F.array([F.lit(i) for i in range(SALT_FACTOR)])
).withColumn("salt_idx", F.explode("salt_array")) \
 .withColumn("salted_key", F.concat(F.col("user_id"), F.lit("_"), F.col("salt_idx")))

# Perform balanced distributed join without stragglers
df_joined = df_left_salted.join(
    df_right_replicated,
    on="salted_key",
    how="inner"
).drop("salted_key", "salt_array", "salt_idx")