Sob o paradigma RDD anterior, eu poderia especificar uma chave e, em seguida, mapear uma operação para os elementos RDD correspondentes a cada chave. Não vejo uma maneira clara de fazer isso com o DataFrame no SparkR a partir da 1.5.1. O que eu gostaria de fazer é algo comodplyr Operação:

new.df <- old.df %>%
  group_by("column1") %>%
  do(myfunc(.))

Atualmente, tenho um SparkFr DataFrame grande do formulário:

            timestamp  value  id
2015-09-01 05:00:00.0  1.132  24
2015-09-01 05:10:00.0  null   24
2015-09-01 05:20:00.0  1.129  24
2015-09-01 05:00:00.0  1.131  47
2015-09-01 05:10:00.0  1.132  47
2015-09-01 05:10:00.0  null   47

Eu classifiquei por,id etimestamp.

Quero agrupar porid, mas não quero agregar. Em vez disso, quero fazer um conjunto de transformações e cálculos em cada grupo - por exemplo, interpolar para preencher NAs (que são gerados quando eucollect DataFrame e depois convertavalue numérico). Eu testei usandoagg, mas enquanto meus cálculos parecem rodar, os resultados não são retornados, porque eu não estou retornando um único valor emmyfunc:

library(zoo)

myfunc <- function(df) {

  df.loc <- collect(df)
  df.loc$value <- as.numeric(df.loc$value)
  df.loc$newparam <- na.approx(df.loc$value, na.rm = FALSE)
  return(df.loc)

  # I also tested return(createDataFrame(sqlContext, df.loc)) here

}

df <- read.df( # some stuff )

grp <- group_by(df, "id")

test <- agg(grp, "myfunc")

15/11/11 18:45:33 INFO scheduler.DAGScheduler: Job 2 finished: dfToCols at NativeMethodAccessorImpl.java:-2, took 0.463131 s
   id
1  24
2  47

Observe que as operações emmyfunc tudo funciona corretamente quando eufilter o DataFrame até um únicoid e execute. Com base no tempo de execução (cerca de 50 segundos por tarefa) e no fato de não haver exceções, acreditomyfunc está realmente sendo executado em todos osids - mas eu preciso da saída!

Qualquer entrada seria apreciada.