The first fix is to avoid the shuffle: broadcast the small side and skew no longer matters (Broadcast Joins). When both sides are large, salting splits each hot key by hand: add a random salt from 0 to N-1 to the big side, replicate every row of the other side N times with each salt value, and join on the key plus the salt. AQE skew join handling (on by default) splits oversized partitions automatically in sort-merge joins. The listing continues Listing 5.6.6.
N = 8 # salt values per hot key
salted_lines = lines.withColumn("salt", (F.rand(seed=7) * N).cast("int"))
salted_catalog = catalog.withColumn("salt", F.explode(F.sequence(F.lit(0), F.lit(N - 1))))
join_tasks("salted", salted_lines.join(salted_catalog, ["book_id", "salt"]))
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true") # the default
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "8m") # 256m
join_tasks("aqe", lines.join(catalog, "book_id"))salted 9 tasks, median 149,526 rows, max 194,021 aqe 7 tasks, median 214,112 rows, max 296,130
Salting cut the largest task from 415,088 to 194,021 rows at the price of eight copies of the catalog and an extra column. AQE split book 1's partition itself (the final plan shows SortMergeJoin(skew=true)), so the largest task is now book 5's. AQE treats a partition as skewed only when it is larger than skewedPartitionFactor (5) times the median and larger than skewedPartitionThresholdInBytes (256 MB); the listing scales the threshold down to suit sample data. Prefer broadcasting, then AQE, then filtering junk keys (a NULL key is the classic hot spot), and salt only what remains, such as skewed aggregations.