Everything in detail about "Shuffle Hash join" in Spark

-

Shuffle Hash Join, as the name indicates works by shuffling both datasets. So the same keys from both sides end up in the same partition or task. Once the data is shuffled, the smallest of the two will be hashed into buckets and a hash join is performed within the partition.

Shuffle Hash Join is different from Broadcast Hash Join because the entire dataset is not broadcasted instead both datasets are shuffled and then the smallest side data is hashed and bucketed and hash joined with the bigger side in all the partitions.

Shuffle Hash Join is divided into 2 phases.

Shuffle phase – both datasets are shuffled

Hash Join phase – smaller side data is hashed and bucketed and hash joined with he bigger side in all the partitions.

Sorting is not needed with Shuffle Hash Joins inside the partitions.

Example

spark.sql.join.preferSortMergeJoin should be set to false and spark.sql.autoBroadcastJoinThreshold should be set to lower value so Spark can choose to use Shuffle Hash Join over Sort Merge Join.

scala> spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 2)scala> spark.conf.set("spark.sql.join.preferSortMergeJoin", "false")scala> spark.conf.get("spark.sql.join.preferSortMergeJoin")res2: String = falsescala> spark.conf.get("spark.sql.autoBroadcastJoinThreshold")res3: String = 2scala> 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 ShuffledHashJoin is used.

scala> dfJoined.queryExecution.executedPlanres4: org.apache.spark.sql.execution.SparkPlan =ShuffledHashJoin [id1#3], [id2#8], Inner, BuildRight:- Exchange hashpartitioning(id1#3, 200):  +- LocalTableScan [id1#3]+- Exchange hashpartitioning(id2#8, 200)   +- LocalTableScan [id2#8]scala> dfJoined.show+---+---+|id1|id2|+---+---+| 20| 20|| 20| 20|| 40| 40|| 40| 40|| 20| 20|| 20| 20|| 20| 20|| 20| 20|| 50| 50|| 30| 30|+---+---+

Internal workings for Shuffle Hash Join

There are 2 phases in a Shuffle Hash Join – Shuffle phase and Hash Join 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.

Hash Join

  1. After the shuffle, Spark picks one side based on the statistics and will hash the side by key in to buckets
  2. In the below example, we have 2 partitions and side 2 is picket for hashing and will be assigned to buckets. There is one bucket in partition 1 with key 20. Partition 2 has 2 buckets 20 and 40 are assigned to bucket 1 and 50 assigned to bucket 2.
  3. Keys from the big dataset will be attempted to match ONLY with the respective buckets. For eg. in partition 1 when the hash value of 101 results in anything other than bucket 1 a match will not be attempted. In partition 2, 401 will only be attempted to match with keys in bucket 1 and not with bucket 1 because 401 is hashed to bucket 2.
  4. Hash join is performed across all partitions after the shuffle.

Sorting is not needed for Shuffle Hash Joins.

When does Shuffle Hash Join work?

  • Faster than a sort merge join since sorting is not involved.
  • Works only for equi joins
  • Works for all join types
  • Works well when a dataset can not be broadcasted but one side of partitioned data after shuffling will be small enough for hash join.

 

When Shuffle Hash Join doesn’t work?

  • Does not works for non-equi joins
  • Does not work with data which are heavily skewed. Let’s say we are joining a sales dataset on the product key. It is possible that the dataset has a disproportionate number of records for a certain product key. Shuffle will result in sending all the records for this product key to a single partition. Hashing all the records for this product key inside a single partition will result in an Out of Memory exception. So Shuffle Hash Join will work for a balanced dataset but not for skewed dataset.

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