There are 2 distinct concepts at play here which explain how an intelligent query execution system works be it Spark or RDBMS.
1.Proving accurate result for the query/execution
A query/execution is parsed into a DAG which represents different execution steps and dependencies between them. The steps can be expressed as a Map
or Reduce
kind of steps. Each independent step is a "stage" and 2 stages are separated by a shuffle boundary.
At no cost these dependencies among stages can be broken, they will run serially (in a given executor).
In this post of mine I have explained how spark executes things in the order provided to provide correct result - Spark withColumn and where execution order
2.Provide that result fast
Within a stage, based on how DAG is defined, certain steps can be parallelized. This is where you see Spark would optimize the execution plan using many mechanisms like - being lazy, running a step before other, catalyst, encoding, full stage code generation, using statistics, predicate push down, columnar access, caching etc. New techniques are added as things evolve. This is where Spark beats Hadoop. In Hadoop you will need to write all optimizations yourself but Spark take care of it behind the scene. The same RDBM works too. I can explain each technique if needed.
The data to be processed is split up among many executors which run the same "stage" on different executors. This is called scalability. as you grow the cluster size (for a large data set) then job would run faster. This behavior is same as Hadoop. The developer is still responsible to some extent to code in a certain way to make sure maximum parallelism is achieved.
Lets see your example
The limit
can't provide accurate result if orderBy
didn't happen 1st. So it will execute in the order orderBy
then limit
. It will never rearrange this order of execution.
val df = spark.createDataset(List(("a","b","c"),("a1","b1","c1"),......).toDF("guitarid","make","model")
df.cache()//without this I was not getting the full plan.
val df1 = df.orderBy("make").limit(1)
df1.show(false)
df1.explain(true)
Plan is as below. Logical plan suggests the order of execution. The physical plan has optimized that execution using a special stage "TakeOrderedAndProject".
== Analyzed Logical Plan ==
guitarid: string, make: string, model: string
GlobalLimit 1
+- LocalLimit 1
+- Sort [make#8 ASC NULLS FIRST], true
+- Project [_1#3 AS guitarid#7, _2#4 AS make#8, _3#5 AS model#9]
+- LocalRelation [_1#3, _2#4, _3#5]
== Optimized Logical Plan ==
GlobalLimit 1
+- LocalLimit 1
+- Sort [make#8 ASC NULLS FIRST], true
+- InMemoryRelation [guitarid#7, make#8, model#9], StorageLevel(disk, memory, deserialized, 1 replicas)
+- LocalTableScan [guitarid#7, make#8, model#9]
== Physical Plan ==
TakeOrderedAndProject(limit=1, orderBy=[make#8 ASC NULLS FIRST], output=[guitarid#7,make#8,model#9])
+- InMemoryTableScan [guitarid#7, make#8, model#9]
+- InMemoryRelation [guitarid#7, make#8, model#9], StorageLevel(disk, memory, deserialized, 1 replicas)
+- LocalTableScan [guitarid#7, make#8, model#9]
If we call limit
before orderBy
, then it maintains same order - limits 1st then sorts to make sure the result is as you expect. It won't give wrong result for performance
val df1 = df.limit(1).orderBy("make")
df1.show(false)
df1.explain(true)
== Analyzed Logical Plan ==
guitarid: string, make: string, model: string
Sort [make#8 ASC NULLS FIRST], true
+- GlobalLimit 1
+- LocalLimit 1
+- Project [_1#3 AS guitarid#7, _2#4 AS make#8, _3#5 AS model#9]
+- LocalRelation [_1#3, _2#4, _3#5]
== Optimized Logical Plan ==
Sort [make#8 ASC NULLS FIRST], true
+- GlobalLimit 1
+- LocalLimit 1
+- InMemoryRelation [guitarid#7, make#8, model#9], StorageLevel(disk, memory, deserialized, 1 replicas)
+- LocalTableScan [guitarid#7, make#8, model#9]
== Physical Plan ==
*(2) Sort [make#8 ASC NULLS FIRST], true, 0
+- *(2) GlobalLimit 1
+- Exchange SinglePartition
+- *(1) LocalLimit 1
+- InMemoryTableScan [guitarid#7, make#8, model#9]
+- InMemoryRelation [guitarid#7, make#8, model#9], StorageLevel(disk, memory, deserialized, 1 replicas)
+- LocalTableScan [guitarid#7, make#8, model#9]
Another example - when you want 2 data frames joined, Spark may choose Hashjoin vs broadcasthashjoin for performance but the end result will be same.
On the other hand if we had code like below. Since these 2 operation depend on separate columns they can execute in any order.
df.withColumn("column10", expression on colum1)
.withColumn("column11", expression on colum2)
Conclusion
I will trust Spark's execution engine to provide accurate result in a performant way. The performance will improve automatically as the execution engine is upgraded, so just stick to Spark's latest syntax.