Understand "Shuffle Sort Merge Join" 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.

We will understand all the above points with examples and elaborate DAGs

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 = truescala> spark.conf.get("spark.sql.autoBroadcastJoinThreshold")res2: String = -1scala> 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.executedPlanres3: 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.

Popular posts from this blog

Window function in PySpark with Joins example using 2 Dataframes (inner join)

-
Input details: #● File has json records #● Each record has fields: #○ user_id #○ card_num #○ merchant #○ category #○ amount #○ ts ### Below analysis to be done Sample data: +------+--------+---------+--------+----------+-------+|amount|card_num| category|merchant| ts|user_id|+------+--------+---------+--------+----------+-------+| 243| C_108| food| M_102|1579532902| U_104|| 699| C_106|cosmetics| M_103|1581759040| U_103|| 228| C_104| children| M_110|1584161986| U_103| Solution: from pyspark.sql.functions import col # this can be done without using window function creditCardData = spark.read.json(“card_transactions.json”) useridMaxSpendDF = df.groupby(‘user_id’).max(‘amount’) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“max(amount)”,”max_amount”) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“user_id”,”m_user_id”) cond = [creditCardData.user_id == useridMaxSpen...
Read more

Complex SQL: fetch the users who logged in consecutively 3 or more times (lead perfect example)

-
From the login_details table, fetch the users who logged in consecutively 3 or more times. Table Name : LOGIN_DETAILS Approach : We need to fetch users who have appeared 3 or more times consecutively in login details table. There is a window function which can be used to fetch data from the following record. Use that window function to compare the user name in current row with user name in the next row and in the row following the next row. If it matches then fetch those records. –Table Structure: drop table login_details; create table login_details( login_id int primary key, user_name varchar(50) not null, login_date date); delete from login_details; insert into login_details values (101, ‘Michael’, current_date), (102, ‘James’, current_date), (103, ‘Stewart’, current_date+1), (104, ‘Stewart’, current_date+1), (105, ‘Stewart’, current_date+1), (106, ‘Michael’, current_date+2), (107, ‘Michael’, cur...
Read more

Credit Card Data Analysis using PySpark (how to use auto broadcast join after disabling it)

-
Input details: #● File has json records #● Each record has fields: #○ user_id #○ card_num #○ merchant #○ category #○ amount #○ ts ### Below analysis to be done Sample data: +------+--------+---------+--------+----------+-------+|amount|card_num| category|merchant| ts|user_id|+------+--------+---------+--------+----------+-------+| 243| C_108| food| M_102|1579532902| U_104|| 699| C_106|cosmetics| M_103|1581759040| U_103|| 228| C_104| children| M_110|1584161986| U_103| Application: Here we will disable auto broadcast and then join 2 data frames and will see execution plan which should use broadcast join Solution: from pyspark.sql import SparkSession from pyspark.sql.functions import broadcast spark = SparkSession.builder.master(‘local[2]’)\ .appName(‘RDD_Methods_Examples’)\ .getOrCreate() print(spark.conf.get(“spark.sql.autoBroadcastJoinThreshold”)) spark.conf.set(“spark.sql.autoBroadcastJoinThreshold”, -1) pr...
Read more