Difference Beetween Window function and OrderBy in Spark-CodePudding

The

CodePudding user response：

Run Explain on both queries, and that will show you the different paths they take. They both look at all the data(I was wrong in my initial answer)

Window sends all the data to 1 node. In this case you also use a where clause that means it uses a shuffle to complete the filtering. This seems to be faster than the implementation used by limit for a large number of items. It's likely faster because of the volume of data. Extra shuffles hurt for small data sets but if optimized correctly on large data sets helps spread the load and reduce the time it takes.

== Physical Plan == *(2) Filter (isnotnull(row_number#66) && (row_number#66 <= 1000000000)) - Window [row_number() windowspecdefinition(date#61 ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS row_number#66], [date#61 ASC NULLS FIRST] - *(1) Sort [date#61 ASC NULLS FIRST], false, 0 - Exchange SinglePartition - Scan hive default.table_persons [people#59, type#60, date#61], HiveTableRelation `default`.`table_persons`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [people#59, type#60, date#61]

"Order by" uses a different implementation and does not appear to perform as well under with 'large takes'. It doesn't use as many shuffles but this seems that it doesn't accomplish the work as fast when there is a large number of items to return.

== Physical Plan == TakeOrderedAndProject(limit=100000000, orderBy=[date#61 ASC NULLS FIRST], output=[people#59,type#60,date#61]) - Scan hive default.table_persons [people#59, type#60, date#61], HiveTableRelation `default`.`table_persons`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [people#59, type#60, date#61]

As an aside, for large skewed data sets it common to add 2 extra shuffles and this does in fact take less time than not having the extra shuffles. (But again would increase the time it takes on small data sets.)

So what does TakeOrderedAndProject actually do? It uses a take and sorts the data on disk for large datasets. (Instead of sorting in memory).

With your window it does do a shuffle which does use ranges to sort the data. There are inferences that further sorting is done in memory giving you the performance tradeoff. (Updated links to inferences below.)

And this is where I think you are getting the payoff. (I'd be interested to know if adding a limit to your existing window speed things up.)

From snippets on reddit and digging issues the code it the are inference that the sort is done in memory and spilt to disk if it's required.

The other piece that I feel provide a lot of performance boost is that you use a where clause. Take pulls back as many items from each partition as you have in the limit clause. (see the implementation above.) This is done only to throw out the items. where doesn't not pull any items back that do not match the filter condition. This movement of data [using limit] is likely where you are getting the real performance degradation in limit.