How 'sort merge' helps in performance while merging 2 tables in spark

-

Apache Spark Join Optimization – 2 Large Tables

Intent should be to minimize shuffling & maximize parallelism.

Minimize Shuffling –
Try filtering the data before shuffle.
Cut down the size as early as possible to minimize shuffling,
do any of the aggregation before only.

Maximize Parallelism –
1. procure the right number of CPU cores.
If you request for 20 executors with 5 cores,
then at the max we can have 100 tasks running in parallel.

2. set the right number of shuffle partitions.
If you have set it to 50,
then you can have at the max 50 parallel tasks.

3. cardinality should be high.
If you have 20 distinct keys,
then at the max 20 tasks will be doing work in parallel.

so if you get 20 containers with 5 cores each,
your shuffle partitions is set to 50 and,
there are 20 distinct keys.
Then you can have 20 tasks running in parallel.

Max Parallelism = Min(total_CPU_Cores, shuffle_Partitions, Cardinality)

4. Avoid partition skew – Consider 20 tasks. 1 of them is overloaded with work.
The completion of job is dependent on slowest performing task.

Answer:

Here intent should be minimize shuffling so it can be achieved by merge-sort join.
Most of the joining algorithms, first perform repartitioning and then joining. So after repartitioning – Merge-sort involves first sorting and then merging, and sorting is done at partition-level only and that too parallelly. So at the time of merging, the spark engine simply has to merge sorted data present in each partition.
A simple concept – sorting a bigger dataset will take longer than sorting a smaller dataset. So first sort and then simply merge.
Also, since Spark 2.3(or 2.4) the property value spark.sql.join.preferSortMergeJoin is true by default

Popular posts from this blog

How to change column name in Dataframe and selection of few columns in Dataframe using Pyspark with example

-
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

What is Garbage collection in Spark and its impact and resolution

-
Image
Garbage Collection Spark runs on the Java Virtual Machine (JVM). Because Spark can store large amounts of data in memory, it has a major reliance on Java’s memory management and garbage collection (GC). Therefore, garbage collection  (GC) can be a major issue that can affect many Spark applications. Common symptoms of excessive GC in Spark are: Slowness of application Executor heartbeat timeout GC overhead limit exceeded error Spark’s memory-centric approach and data-intensive applications make it a more common issue than other Java applications. Thankfully, it’s easy to diagnose if your Spark application is suffering from a GC problem. The Spark UI marks executors in red if they have spent too much time doing GC. Spark executors are spending a significant amount of CPU cycles performing garbage collection. This can be determined by looking at the “Executors” tab in the Spark application UI. Spark will mark an executor in  red  if the executor has spent more than 10% of the time in gar...
Read more

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