How does merge-sort join work in Spark and why can it throw OOM?

user_706354 0
Viewed 915

I want to understand the concept of merge-sort join in Spark in depth. I understand the overall idea: this is the same approach as in merge sort algorithm: Take 2 sorted datasets, compare first rows, write smallest one, repeat. I also understand how I can implement distributed merge sort.

But I cannot get how it is implemented in Spark with respect to concepts of partitions and executors.

Here is my take.

  1. Given I need to join 2 tables A and B. Tables are read from Hive via Spark SQL, if this matters.
  2. By default Spark uses 200 partitions.
  3. Spark then will calculate join key range (from minKey(A,B) to maxKey(A,B) ) and split it into 200 parts. Both datasets to be split by key ranges into 200 parts: A-partitions and B-partitions.
  4. Each A-partition and each B-partition that relate to same key are sent to same executor and are sorted there separatelt from each other.
  5. Now 200 executors can join 200 A-partitions with 200 B-partitions with guarantee that they share same key range.
  6. The join happes via merge-sort algo: take smallest key from A-partition, compare with smallest key from B-partition, write match, or iterate.
  7. Finally, I have 200 partitions of my data which are joined.

Does it make sense?

Issues: Skewed keys. If some key range comprises 50% of dataset keys, some executor would suffer, because too many rows would go to the same partition. It can even fail with OOM, while trying to sort too big A-partition or B-partition in memory (I cannot get why Spark cannot sort with disk spill, as Hadoop does?..) Or maybe it fails because it tries to read both partitions into memory for joining?

So, this was my guess. Could you please correct me and help to understand the way Spark works?

apache-spark
2 Answers

This is a common problem with joins on MPP databases and Spark is no different. As you say, to perform a join, all the data for the same join key value must be colocated so if you have a skewed distribution on the join key, you have a skewed distribution of data and one node gets overloaded.

If one side of the join is small you could use a map side join. The Spark query planner really ought to do this for you but it is tunable - not sure how current this is but it looks useful.

Did you run ANALYZE TABLE on both tables?

If you have a key on both sides that won't break the join semantics you could include that in the join.

user_47033320

why Spark cannot sort with disk spill, as Hadoop does?

Spark merge-sort join does spill to disk. Taking a look at Spark SortMergeJoinExec class, it uses ExternalAppendOnlyUnsafeRowArray which is described as:

An append-only array for UnsafeRows that strictly keeps content in an in-memory array until numRowsInMemoryBufferThreshold is reached post which it will switch to a mode which would flush to disk after numRowsSpillThreshold is met (or before if there is excessive memory consumption)

This is consistent with the experience of seeing tasks spilling to disk during a join operation from the Web UI.

why [merge-sort join] can throw OOM?

From the Spark Memory Management overview:

Spark’s shuffle operations (sortByKey, groupByKey, reduceByKey, join, etc) build a hash table within each task to perform the grouping, which can often be large. The simplest fix here is to increase the level of parallelism, so that each task’s input set is smaller.

i.e. in the case of join, increase spark.sql.shuffle.partitions to reduce the size of the partitions and the resulting hash table and correspondingly reduce the risk of OOM.

user_18593700
Related