11

Let's say I have the following pipeline:

df.orderBy('foo').limit(10).show()

Here we can see that the orderBy instruction comes first, so all rows of the dataframe should be sorted before the limit instruction be executed. I found myself thinking if the Spark does some "reorganization" inside the pipeline in order to improve performace (for example, executing the limit instruction before the orderBy). Does spark do that?

5 Answers 5

10

Your assumption is correct. Spark executes sort and then limit on each partition before merging/collecting the results as we will see next.

An orderBy followed by limit will cause the next calls:

By looking into the TakeOrderedAndProjectExec:doExecute() method we will first meet the next code:

protected override def doExecute(): RDD[InternalRow] = {
    val ord = new LazilyGeneratedOrdering(sortOrder, child.output)
    val localTopK: RDD[InternalRow] = {
      child.execute().map(_.copy()).mapPartitions { iter =>
        org.apache.spark.util.collection.Utils.takeOrdered(iter, limit)(ord)
      }
    }

......

Here we can see that the localTopK is populated by getting topK first records from each sorted partition. That means that Spark tries to push-down the topK filter as soon as possible at partition level.

The next lines:

....

val shuffled = new ShuffledRowRDD(
      ShuffleExchangeExec.prepareShuffleDependency(
        localTopK,
        child.output,
        SinglePartition,
        serializer,
        writeMetrics),
      readMetrics)
    shuffled.mapPartitions { iter =>
      val topK = org.apache.spark.util.collection.Utils.takeOrdered(iter.map(_.copy()), limit)(ord)
      if (projectList != child.output) {
        val proj = UnsafeProjection.create(projectList, child.output)
        topK.map(r => proj(r))
      } else {
        topK
      }
    }

Will generate the final ShuffledRowRDD from all partitions which will contain the final topK sorted records composing the final result of limit.

Example

Let's illustrate this through an example. Consider the dataset with the range 1,2,3...20 which is partitioned into two parts. The first one contains the odd numbers when the second one the even numbers as shown next:

-----------   -----------
|   P1    |   |   P2    | 
-----------   -----------
|   1     |   |   2     |
|   3     |   |   4     |
|   5     |   |   6     |
|   7     |   |   8     |
|   9     |   |   10    |
|  ....   |   |  ....   |
|   19    |   |   20    |
-----------   -----------

when df.orderBy(...).limit(5) is executed Spark will get top 5 sorted records from each partition aka 1-9 for the 1st one and 2-10 for the 2nd one. Then it will merge and sort them aka sequence 1,2,3,4,5..10. Finally it will get the top 5 records generating the final list 1,2,3,4,5.

Conclusion

Spark leverages all the available information when it comes to orderBy followed by limit by omitting to process the whole dataset but only the first topK rows. As @ShemTov already mentioned there is no need to call limit before orderBy since 1st that would return an invalid dataset and 2nd because Spark does all the necessary optimisations internally for you.

4

Spark does optimization when need, but in your case it cant do the limit before orderBy because you`ll get uncorrect results.

This code mean i want spark to order all rows on foo column, and then give me the top 10.

3

Yes! Spark does 'rule-based' optimizations in instructions before execution. Spark can do this because all the transformations (.select(),.orderBy(), .limit() etc) are lazy.

In few words, Spark context follows the next procedure

  • Unresolved Logical plan: Firstly, Spark context creates instructions without using metadata. For example if in the plan there is a column name that is not exist, the plan will not have problem, because it's unresolved.

  • Logical plan : In the next step, Spark verify the the created instructions with the data of the "Catalog" (e.g table name, column names, semantics)

  • Optimized logical plan: At this stage, the instructions will change due to "Catalyst Optimizer"!

  • Physical plans: At this final stage we have the final instructions, the instructions which will create the execution code for the JVM.

Example:

I used .explain() in order to see the final physical plan.

If i run this code: df.orderBy('foo').limit(20).limit(5).explain(), the physical plan will be:

== Physical Plan ==
TakeOrderedAndProject(limit=5, orderBy=[foo#0L ASC NULLS FIRST], output=[foo#0L])
+- Scan ExistingRDD[foo#0L]

Hmm interesting.. Spark instructions after optimization removed the .limit(20) because it's useless. Spark order and then project for each partition in order to do this task in parallel. Finally, will merge the results and show the final top 5 records.

About your example in your question

In this case: df.orderBy('foo').limit(10).show()

If you run this tranformations with .show() action (the default number of lines in show is 20), so the the Spark will limit the result in 10 records ( Because 10 < 20) with the same procedure as I explained above(apply TakeOrderedAndProject method).

2
+25

Simply yes it does, but it doesn't change the result in any case. Thats why we called it optimization.

Spark gives us two operations for performing any problem.

When we do a transformation on any RDD, it gives us a new RDD. But it does not start the execution of those transformations. The execution is performed only when an action is performed on the new RDD and gives us a final result.

So once you perform any action on an RDD, Spark context gives your program to the driver.

The driver creates the DAG (directed acyclic graph) or execution plan (job) for your program. Once the DAG is created, the driver divides this DAG into a number of stages. These stages are then divided into smaller tasks and all the tasks are given to the executors for execution.

The Spark driver is responsible for converting a user program into units of physical execution called tasks. At a high level, all Spark programs follow the same structure. They create RDDs from some input, derive new RDDs from those using transformations, and perform actions to collect or save data. A Spark program implicitly creates a logical directed acyclic graph (DAG) of operations.

When the driver runs, it converts this logical graph into a physical execution plan.

2

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.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.