SparkR和dplyr:使用gapply的窗口函数计数()

Question

我试着用"gapply“来实现对星火的简单查询，但是会遇到麻烦。

这个代码很好用。

library(SparkR)
library(dplyr)
df <- createDataFrame(iris)
createOrReplaceTempView(df, "iris")
display(SparkR::sql("SELECT *, COUNT(*) OVER(PARTITION BY Species) AS RowCount FROM iris"))

但我无法通过实现这一点

display(df %>% SparkR::group_by(df$Species) 
           %>% gapply(function(key, x) { y <- data.frame(x, SparkR::count()) }, 
               "Sepal_Length double, Sepal_Width double, Petal_Length double, Petal_Width double, Species string, RowCount integer"))

返回错误

SparkException: r意外退出。由: EOFException: org.apache.spark.SparkException:由于阶段失败导致作业中止:阶段235.0中的任务0失败4次，最近的失败: 235.0阶段中丢失任务0.3 (TID 374) (10.150.202.5执行器1)：org.apache.spark.SparkException: r意外退出。R工作者产生错误: Error in (函数(classes，fdef，mtable) )：

无法为函数“计数”找到用于签名“缺失”调用的继承方法:计算.computeFunc -> data.frame -> ->执行暂停

是否可以使用dplyr的管道使用dplyr的管道来实现窗口函数"count“

Answer 1

只是一个小错误，您应该使用base::nrow函数而不是在gapply中使用SparkR::count。

display(df %>% SparkR::group_by(df$Species) 
           %>% gapply(function(key, x) { y <- data.frame(x, nrow(x)) }, 
               "Sepal_Length double, Sepal_Width double, Petal_Length double, Petal_Width double, Species string, RowCount integer"))

这是如何通过SparkR API使用SparkR::windowPartitionBy函数实现的，这里不需要创建任何UDF -

(
  df %>%
  SparkR::select(
    c(
      SparkR::columns(df),
      SparkR::over(
        SparkR::count(SparkR::lit(1)), 
        SparkR::windowPartitionBy(SparkR::column("Species"))
      ) %>% SparkR::alias("RowCount")
    )
  ) %>%
  display()
)