在Sparklyr中按组计算分位数

Question

我在Spark中有一个数据框，并想按特定列分组后计算0.1分位数。

例如：

> 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

我的第一个想法是使用group_by和summarise，但是显然分位数功能未在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

如此处建议的那样，使用sdl_quantile并不算运气：https : //github.com/rstudio/sparklyr/issues/204。请注意，我刚刚升级了sparklyr，并从github运行了版本0.7.0-9004。

> 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有效-但这不是我感兴趣的：

> sdf_quantile(diamonds_sdl, "carat", 0.1)
 10%
0.31

我尝试了其他方法。

首先是使用spark_apply。但是，它似乎在我的安装中无法正常工作。另一个运行返回一个错误，指出该节点中未安装“ Rscript”。但是，由于我没有管理员权限，因此无法真正解决此问题。

> 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)

第二种方法是按以下方式使用collect_list：Sparklyr：使用group_by，然后将组中行中的字符串连接起来

> 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

Answer 1

对于分组数据，最好的选择是percentile_approx：

diamonds_sdl %>% group_by(color) %>% summarise(q1 = percentile_approx(carat, .1))

diamonds_sdl %>% group_by(color) %>% summarise(q1 = percentile_approx(carat, .1))

但是，这需要Spark，并且启用了Hive支持，并且效率不如内置approxQuantile。