How does Shuffle Sort Merge Join work in Spark?

Shuffle Sort Merge Join, as the name indicates, involves a sort operation. Shuffle Sort Merge Join has 3 phases.

Shuffle Phase – both datasets are shuffled

Sort Phase – records are sorted by key on both sides

Merge Phase – iterate over both sides and join based on the join key.

Shuffle Sort Merge Join is preferred when both datasets are big and can not fit in memory – with or without shuffle.

Do you like us to send you a 47 page Definitive guide on Spark join algorithms? ===> Send me the guide

Example

spark.sql.join.preferSortMergeJoin  by default is set to true as this is preferred when datasets are big on both sides. 

Spark will pick Broadcast Hash Join if a dataset is small. In our case both datasets are small so to force a Sort Merge join we are setting spark.sql.autoBroadcastJoinThreshold  to -1 and this will disable Broadcast Hash Join.

scala> spark.conf.get("spark.sql.join.preferSortMergeJoin")
res1: String = true

scala> spark.conf.get("spark.sql.autoBroadcastJoinThreshold")
res2: String = -1

scala> val data1 = Seq(10, 20, 20, 30, 40, 10, 40, 20, 20, 20, 20, 50)
data1: Seq[Int] = List(10, 20, 20, 30, 40, 10, 40, 20, 20, 20, 20, 50)

scala> val df1 = data1.toDF("id1")
df1: org.apache.spark.sql.DataFrame = [id1: int]

scala> val data2 = Seq(30, 20, 40, 50)
data2: Seq[Int] = List(30, 20, 40, 50)

scala> val df2 = data2.toDF("id2")
df2: org.apache.spark.sql.DataFrame = [id2: int]

scala> val dfJoined = df1.join(df2, $"id1" === $"id2")
dfJoined: org.apache.spark.sql.DataFrame = [id1: int, id2: int]

When we see the plan that will be executed, we can see that SortMergeJoin is used.

scala> dfJoined.queryExecution.executedPlan

res3: org.apache.spark.sql.execution.SparkPlan =
*(3) SortMergeJoin [id1#3], [id2#8], Inner
:- *(1) Sort [id1#3 ASC NULLS FIRST], false, 0
:  +- Exchange hashpartitioning(id1#3, 200)
:     +- LocalTableScan [id1#3]
+- *(2) Sort [id2#8 ASC NULLS FIRST], false, 0
   +- Exchange hashpartitioning(id2#8, 200)
      +- LocalTableScan [id2#8]


scala> dfJoined.show

+---+---+
|id1|id2|
+---+---+
| 20| 20|
| 20| 20|
| 20| 20|
| 20| 20|
| 20| 20|
| 20| 20|
| 40| 40|
| 40| 40|
| 50| 50|
| 30| 30|
+---+---+

Stages involved in Shuffle Sort Merge Join

As we can see below a shuffle is needed with Shuffle Hash Join. First dataset is read in Stage 0 and the second dataset is read in Stage 1. Stage 2 below represents the shuffle.

Inside Stage 2 records are sorted by key and then merged to produce the output.

Internal workings for Shuffle Sort Merge Join

There are 3 phases in a Sort Merge Join – shuffle phase, sort phase and merge phase.

Shuffle phase

Data from both datasets are read and shuffled. After the shuffle operation, records with the same keys from both datasets will end up in the same partition after the shuffle. Note that the entire dataset is not broadcasted with this join. This means the dataset in each partition will be in a manageable size after the shuffle.

Sort phase 

Records on both sides are sorted by key. Hashing and bucketing are not involved with this join.

Merge phase

A join is performed by iterating over the records on the sorted dataset. Since the dataset is sorted the merge or the join operation is stopped for an element as soon as a key mismatch is encountered. So a join attempt is not performed on all keys.

For eg. in partition 2 when keys are attempted to match for 201 records are iterated through on the other side up until 402 is reached as the records are sorted there is no need to iterate through the entire records on the other side.

When does Shuffle Sort Merge Join work?

• Works only on equi joins
• Work on all join type
• Works on big datasets
• Both shuffle and sort on keys are involved with this join.

When Shuffle Sort Merge Join doesn’t work?

• Doesn’t work on non equi joins.
• Both shuffle and sort are expensive operations. Use this join when a broadcast hash and shuffle hash joins are not possible.

Interested in learning about Broadcast Hash Join in Spark? – Click here.