How to calculate Percentile of column in a DataFrame in spark?

Question

I am trying to calculate percentile of a column in a DataFrame? I cant find any percentile_approx function in Spark aggregation functions.

For e.g. in Hive we have percentile_approx and we can use it in the following way

hiveContext.sql("select percentile_approx("Open_Rate",0.10) from myTable); 

But I want to do it using Spark DataFrame for performance reasons.

Sample data set

|User ID|Open_Rate|
------------------- 
|A1     |10.3     |
|B1     |4.04     |
|C1     |21.7     |
|D1     |18.6     |

I want to find out how many users fall into 10 percentile or 20 percentile and so on. I want to do something like this

df.select($"id",Percentile($"Open_Rate",0.1)).show

Solution 1

Since Spark2.0, things are getting easier,simply use this function in DataFrameStatFunctions like :

df.stat.approxQuantile("Open_Rate",Array(0.25,0.50,0.75),0.0)

There are also some useful statistic functions for DataFrame in DataFrameStatFunctions.

Solution 2

SparkSQL and the Scala dataframe/dataset APIs are executed by the same engine. Equivalent operations will generate equivalent execution plans. You can see the execution plans with explain.

sql(...).explain
df.explain

When it comes to your specific question, it is a common pattern to intermix SparkSQL and Scala DSL syntax because, as you have discovered, their capabilities are not yet equivalent. (Another example is the difference between SQL's explode() and DSL's explode(), the latter being more powerful but also more inefficient due to marshalling.)

The simple way to do it is as follows:

df.registerTempTable("tmp_tbl")
val newDF = sql(/* do something with tmp_tbl */)
// Continue using newDF with Scala DSL

What you need to keep in mind if you go with the simple way is that temporary table names are cluster-global (up to 1.6.x). Therefore, you should use randomized table names if the code may run simultaneously more than once on the same cluster.

On my team the pattern is common-enough that we have added a .sql() implicit to DataFrame which automatically registers and then unregisters a temp table for the scope of the SQL statement.

Comments

[zero323]

There is no performance difference between using SQL query and DataFrame - both use the same execution engine.