TopolearnMost slow Spark stages are joins and skew, and AQE plus the right hints fix the majority of them.
Spark picks a physical join strategy:
- Broadcast hash join: the small side is sent to every executor; no shuffle of the big side. Chosen when a side is below
spark.sql.autoBroadcastJoinThreshold(10 MB by default) or hinted withF.broadcast(df). - Shuffle hash join: both sides shuffled by key; the smaller side of each partition is built into a hash table.
- Sort-merge join: both sides shuffled and sorted, then merged. The scalable default for two large tables.
- Broadcast nested loop and cartesian: used for non-equi joins; potentially very slow.
Hints let you choose: df.hint("broadcast"), "merge", "shuffle_hash", or SQL /*+ BROADCAST(t) */.
Adaptive Query Execution (AQE, on by default) re-optimizes at stage boundaries using real runtime statistics:
- Coalescing shuffle partitions: merges many tiny post-shuffle partitions into ones of a sensible size (
advisoryPartitionSizeInBytes, 64 MB by default), so the 200-partition default rarely matters. - Switching join strategy: converts a sort-merge join into a broadcast join when a side turns out to be small after filtering.
- Skew join handling: splits partitions much larger than the median into smaller tasks.
For skew AQE can't fix, such as a giant null or default key, filter it out, handle it separately, or salt the key: add a random suffix on the big side and replicate the small side across the suffixes.
Example
from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.master("local[2]").getOrCreate()
def plan(df):
return df._jdf.queryExecution().executedPlan().toString()
facts = spark.range(200_000).withColumn("k", F.col("id") % 1000)
dims = spark.range(1000).withColumnRenamed("id", "k").withColumn("name", F.concat(F.lit("n"), "k"))
auto = facts.join(dims, "k") # dims is small: broadcast chosen
print("BroadcastHashJoin" in plan(auto)) # True
merged = facts.join(dims.hint("merge"), "k") # force a sort-merge join
print(len(merged.collect())) # 200000 (run it so AQE can re-plan)
p = plan(merged)
print("SortMergeJoin" in p, "isFinalPlan=true" in p) # True True
print("AQEShuffleRead coalesced" in p)
# True: AQE merged the 200 tiny shuffle partitions after seeing real sizesRemember
- Broadcast hash, shuffle hash and sort-merge joins each fit different sizes
- AQE coalesces partitions, switches joins and splits skew at runtime
- Salt or isolate extreme hot keys AQE can't fix
Watch out for
- Forcing a broadcast of a table that is large in memory and crashing executors or the driver.