How does Spark SQL optimize joins? What are the optimization tricks for joins?

How does Spark SQL optimize joins? What are the optimization tricks for joins?


Lets try to understand how spark 2.0 works for DataFrame API Being a DataFrame, spark has knowledge about the structure of the data.

While applying join we always have to think that what type of join I should do and which will  be more efficient and optimize to process the data smoothly?

 When joining big table to small table, it is a good idea to broadcasting the smaller table?

 However when joining big table to big table , what optimization tricks are there? Does sorting help here ? would spark do the sorting internally? When should I repartition the data?


So Answer is Yes!
We can optimize the joins in Spark SQL for better performance and process data smoothly :

As we all now that Spark SQL comes with JoinSelection execution planning strategy that translates a logical join to one of the supported join physical operators (per join physical operator selection requirements).


There are 6 different types of physical join operators which can be apply for optimization:

1.  BroadcastHashJoinExec : when left or right join side can be broadcast (i.e. smaller than  spark.sql.autoBroadcastJoinThreshold  which is 10M by default.)
2. ShuffledHashJoinExec : when  spark.sql.join.preferSortMergeJoin  is disabled and it's possible to build hash maps for left or right join side (among the requirements).
3. SortMergeJoinExec :  when left join keys are "orderable".
4. BroadcastNestedLoopJoinExec :  when there are no joining keys and left or right join side could be broadcast.
5. CartesianProductExec :  when it's inner or cross join with no join condition.
6. BroadcastNestedLoopJoinExec : when no other have matched.


So now we can see there's a lot of theory to digest to "what optimization tricks are there".
1. Does sorting help? 
Yes. See SortMergeJoinExec operator.
2. would spark do the sorting internally?
Yes, It will try, but humans can (still?) do wonders.
3. When should I repartition the data?
Always if we can and know that pruning can help. That can reduce the number of rows to deal with and effectively allows for  BroadcastHashJoinExec   over  ShuffledHashJoinExec   or the others.

I also think that repartitioning the data can be of particular help with cost-based optimization where table pruning could reduce the number of columns and rows, and in turn the table size and the cost of one join over others in general.


Comments

Post a Comment

Popular posts from this blog

Transformations and Actions in Spark

Image
Hello Friends, Today I’ll try to explain about Transformations and Actions in Spark. We know that Basically, Spark  RDD supports two types of Operations :          ·    Transformations          ·    Actions Transformations : Transformations are kind of operations which will transform our RDD data from one form to another form, and when we apply this operation on any RDD, we get a new RDD with transformed data   as RDDs in Spark are immutable. Operations like map, flatMap, filter, join, union are transformations. Now there is a point to be noted here and that is when we apply the transformation on any RDD it will not perform the operation immediately. It creates a DAG(Directed Acyclic Graph) using the applied operations, source RDD and functions used for transformation. And it keeps on building this graph using the references till we apply any action operation on the last lined up RDD. That is why the transformation in Spark are lazy. After the transformatio
Read more

How to Convert a Spark DataFrame to Map in Scala

Image
Hi Friends, In this post, I am going to explain that how we can convert Spark DataFrame to Map. Input DataFrame : Output map : Below is the explained code with all the steps along with its Output.   import org.apache.spark.SparkConf import org.apache.spark.sql.{Row, SparkSession} import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType} object ConvertDFToMap {   def main(args: Array[String]) {     //Creating SparkSession     lazy val conf = new SparkConf().setAppName("DataFrame-to-Map").set("spark.default.parallelism", "1")       .setIfMissing("spark.master", "local[*]")     lazy val sparkSession = SparkSession.builder().config(conf).getOrCreate()     // Creating Seq of Raw Data to create DataFrame to test     val rawData = Seq(       Row(1, "Anamika", "Uttar Pradesh"),       Row(2, "Ananya", "Delhi"),       Row(3, "Ambika",
Read more

How to Handle and Convert DateTime format in Spark-Scala.

Image
Hi Friends, In this post, I'd like to explain that how to handle and convert the date format in expected date format. Below I'm providing few examples : 1. Change Date format to yyyy-mm-dd from dd-mm-yyyy or mm-dd-yyyy : Input Date :  03-04-2020T15:26:51+05:30 Converted Output : 2020-04-03T15:26:051 Code to get the same output : def changeDateFormat (inputDateString: String ) = { var convertedDateFormat = "" try { val inputDateFormat: SimpleDateFormat = new SimpleDateFormat( "dd-MM-yyyy'T'HH:mm:sss" ) val date = inputDateFormat.parse(inputDateString) val df = new SimpleDateFormat( "yyyy-MM-dd'T'HH:mm:sss" ) convertedDateFormat = df.format(date) } catch { case ex: Exception => { "Exception is : " + ex } logger .info( "[HandleDateTimeUtils] Exception while converting time to 'yyyy-MM-dd'T'HH:mm:sss' Format. Input is: " + inputDateString , ex.getMessag
Read more