Spark 2.3中Planner为何默认偏好Sort Merge Join而非Shuffled Hash Join？

Great question—this cuts to how Spark balances performance, reliability, and predictability in distributed join operations. Let’s unpack each part of your query:

1. Why Sort Merge Join (SMJ) is the Preferred Choice #

Spark’s planner leans into SMJ for three core reasons:

• Stability first: Shuffled Hash Join (SHJ) relies on building an in-memory hash table for one of the joined tables. If that table is larger than expected (or hits data skew), the hash table spills to disk—leading to massive performance drops or even OutOfMemoryErrors. SMJ’s sort-then-merge workflow is far more predictable: it processes data in partitioned chunks, so even with large datasets, performance degrades gradually rather than crashing.
• Lower memory sensitivity: SHJ requires enough memory to hold the entire hash table (or at least a large partition of it) for the smaller table. SMJ only needs memory to sort individual partitions, which is a much lighter load. This makes SMJ more resilient to varying memory configurations in production clusters.
• Better skew handling: While both joins suffer from data skew, SMJ’s sorting phase is easier to mitigate (e.g., via salted partitioning to split skewed keys). SHJ, by contrast, can get stuck on a single oversized hash partition that cripples a single task’s performance.

2. Why spark.sql.join.preferSortMergeJoin is an Internal, Default-On Property #

Spark’s team designed this as an internal property to protect users from unintended performance issues:

• SMJ is the "safe default": Most users don’t have the expertise to judge when SHJ is appropriate. Enabling SMJ by default ensures that even less experienced teams get consistent, reliable join performance without manual tuning.
• Prevent misconfiguration: Making it internal discourages casual toggling—SHJ only outperforms SMJ in narrow scenarios (e.g., joining a tiny in-memory table with a large one). Exposing this as a public setting might lead users to switch it on without understanding the tradeoffs, leading to unstable jobs.

3. Key Shortcomings of Shuffled Hash Join #

SHJ’s limitations make it a niche choice:

• Memory dependency: Its performance plummets if the hash table can’t fit in memory. Disk spills add enormous overhead, often making SHJ slower than SMJ in these cases.
• Unpredictable performance: Small changes in data distribution (like sudden skew) can turn a fast SHJ into a job that hangs or fails. SMJ’s performance curve is far more consistent.
• Hash collision risks: While rare, hash collisions degrade hash table efficiency—something SMJ (which relies on sorted order) never has to deal with.
• Niche use case only: SHJ only shines when one table is small enough to fit entirely in memory. For most real-world distributed joins (where both tables are large), SMJ is better.

4. Distributed vs. Algorithm-Inherent Limitations #

This is a mix of both, but Spark’s distributed nature amplifies SHJ’s inherent flaws:

• Algorithm-inherent: Even in a single machine, SHJ depends on sufficient memory for the hash table. But in a distributed cluster:
  • Executors have limited per-task memory, so the threshold for "small enough" is much lower.
  • Disk spills in distributed environments are far more costly (cross-node IO vs. local disk).
• Spark-specific: Data skew is far more impactful in distributed systems—one skewed key can leave a single task processing 90% of the data, which SHJ handles poorly compared to SMJ’s partitioned sorting approach.

5. How preferSortMergeJoin Works in JoinSelection #

In Spark’s JoinSelection planning rule, SMJ takes priority unless explicitly disabled. Here’s a simplified look at the logic:

case join @ Join(left, right, joinType, condition, hint) =>
  // First try Broadcast Hash Join (if a small table can be broadcast)
  broadcastJoin(join).orElse {
    // Check if SMJ is preferred (default: yes)
    if (conf.preferSortMergeJoin) {
      sortMergeJoin(join).orElse(shuffledHashJoin(join))
    } else {
      shuffledHashJoin(join).orElse(sortMergeJoin(join))
    }
  }

SHJ is only selected if:

• preferSortMergeJoin is turned off, or
• SMJ isn’t feasible (e.g., join keys aren’t sortable—though this is rare in Spark), and
• One of the tables is small enough to build an in-memory hash table.

内容的提问来源于stack exchange，提问作者Jacek Laskowski