I have a dataframe in Spark, and would like to calculate the 0.1 quantile after grouping by a specific column.
For example:
> library(sparklyr)
> library(tidyverse)
> con = spark_connect(....)
> diamonds_sdl = copy_to(con, diamonds)
> diamonds
# Source: table<diamonds> [?? x 10]
# Database: spark_connection
carat cut color clarity depth table price x y z
<dbl> <chr> <chr> <chr> <dbl> <dbl> <int> <dbl> <dbl> <dbl>
1 0.230 Ideal E SI2 61.5 55.0 326 3.95 3.98 2.43
2 0.210 Premium E SI1 59.8 61.0 326 3.89 3.84 2.31
3 0.230 Good E VS1 56.9 65.0 327 4.05 4.07 2.31
4 0.290 Premium I VS2 62.4 58.0 334 4.20 4.23 2.63
5 0.310 Good J SI2 63.3 58.0 335 4.34 4.35 2.75
6 0.240 Very Good J VVS2 62.8 57.0 336 3.94 3.96 2.48
7 0.240 Very Good I VVS1 62.3 57.0 336 3.95 3.98 2.47
8 0.260 Very Good H SI1 61.9 55.0 337 4.07 4.11 2.53
9 0.220 Fair E VS2 65.1 61.0 337 3.87 3.78 2.49
10 0.230 Very Good H VS1 59.4 61.0 338 4.00 4.05 2.39
My first thought was to use group_by and summarise, but apparently the quantile function is not implemented in sparklyr:
> diamonds_sdl %>% group_by(color) %>% summarise(q1=quantile(carat, .1))
Error: org.apache.spark.sql.AnalysisException: Undefined function: 'QUANTILE'. This function is neither a registered temporary function nor a permanent function registered in the database 'tsci
'.; line 1 pos 16
No luck using sdl_quantile, as suggested here: https://github.com/rstudio/sparklyr/issues/204 . Note that I've just upgraded sparklyr, and running version 0.7.0-9004 from github.
> diamonds_sdl %>% group_by(color) %>% summarise(q1=sdf_quantile(carat, .1))
Error: org.apache.spark.sql.AnalysisException: Undefined function: 'SDF_QUANTILE'. This function is neither a registered temporary function nor a permanent function registered in the database '
tsci'.; line 1 pos 16
sdf_quantile works if I want to calculate the quantile on the whole column - but this is not what I am interested in:
> sdf_quantile(diamonds_sdl, "carat", 0.1)
10%
0.31
I've tried to other approaches.
The first is to use spark_apply. However it seems that it doesn't work properly in my installation. Another run returned an error saying that "Rscript" was not installed in the node. However, I cannot really fix this problem as I do not have admin rights.
> spark_apply(diamonds_sdl, function(x) quantile(x, 0.1))
Error: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 35.0 failed 4 times, most recent failure: Lost task 0.3 in stage 35.0 (TID 735, myserver
.com, executor 393): java.lang.Exception: sparklyr worker rscript failure with status 255, check worker logs for details.
at sparklyr.Rscript.init(rscript.scala:98)
at sparklyr.WorkerRDD$$anon$2.run(rdd.scala:95)
The second approach was to use collect_list as in here: Sparklyr: Use group_by and then concatenate strings from rows in a group
# This one almost works
> diamonds_sdl %>% group_by(color) %>% summarise(q1=paste(collect_list(as.character(carat))))
# Source: lazy query [?? x 2]
# Database: spark_connection
color
<chr>
1 F
2 E
3 D
4 J
5 G
6 I
7 H
q1
<chr>
1 0.22 0.23 0.23 0.23 0.23 0.29 0.24 0.26 0.7 0.96 0.81 0.8 0.73 0.73 0.8 0.8 …
2 0.23 0.21 0.23 0.22 0.2 0.32 0.23 0.23 0.23 0.23 0.25 0.22 0.24 0.26 0.26 0.…
3 0.23 0.23 0.26 0.26 0.26 0.22 0.3 0.3 0.3 0.24 0.26 0.26 0.26 0.75 0.71 0.61…
4 0.31 0.24 0.3 0.23 0.31 0.3 0.3 0.3 0.31 0.31 0.3 0.33 0.3 1.17 1.05 1.05 1.…
5 0.23 0.23 0.28 0.31 0.31 0.24 0.7 0.78 0.74 0.75 0.75 0.8 0.74 0.71 0.64 0.7…
6 0.29 0.24 0.3 0.3 0.24 0.33 0.33 0.32 0.3 0.3 0.3 0.3 0.35 0.42 0.32 0.38 0.…
# but here is the error
> diamonds_sdl %>% group_by(color) %>% summarise(q1=quantile(as.numeric(paste(collect_list(as.character(carat))))))
Error: org.apache.spark.sql.AnalysisException: Undefined function: 'QUANTILE'. This function is neither a registered temporary function nor a permanent function registered in the database 'tsci
'.; line 1 pos 16