Как использовать применять, чтобы найти максимум в каждой группе SparkR

Question

У меня есть следующий искровой DataFrame:

agent_product_sale=data.frame(agent=c('a','b','c','d','e','f','a','b','c','a','b'), 
         product=c('P1','P2','P3','P4','P1','p1','p2','p2','P2','P3','P3'), 
         sale_amount=c(1000,2000,3000,4000,1000,1000,2000,2000,2000,3000,3000)) 

RDD_aps=createDataFrame(sqlContext,agent_product_sale) 

    agent product sale_amount 
1  a  P1  1000 
2  b  P1  1000 
3  c  P3  3000 
4  d  P4  4000 
5  d  P1  1000 
6  c  P1  1000 
7  a  P2  2000 
8  b  P2  2000 
9  c  P2  2000 
10  a  P4  4000 
11  b  P3  3000 

Мне нужно сгруппировать Спарк DataFrame агентом и для каждого агента найти продукт с самым высоким sale_amount

 agent most_expensive 
     a   P4   
     b   P3     
     c   P3   
     d   P4   

Я использую следующий код, но он будет возвращать максимальное sale_amount для каждого агента

schema <- structType(structField("agent", "string"), 
structField("max_sale_amount", "double")) 

result <- gapply(
RDD_aps, 
c("agent"), 
function(key, x) { 
y <- data.frame(key,max(x$sale_amount), stringsAsFactors = FALSE) 
}, schema) 

Answer 1

с tapply() или агрегатом() вы можете найти максимальное значение в группе

agent_product_sale=data.frame(agent=c('a','b','c','d','e','f','a','b','c','a','b'), 
     +        product=c('P1','P2','P3','P4','P1','p1','p2','p2','P2','P3','P3'), 
     +        sale_amount=c(1000,2000,3000,4000,1000,1000,2000,2000,2000,3000,3000)) 


tapply(agent_product_sale$sale_amount,agent_product_sale$agent, max) 
       a b c d e f 
      3000 3000 3000 4000 1000 1000 



aggregate(agent_product_sale$sale_amount,by=list(agent_product_sale$agent), max) 
      Group.1 x 
     1  a 3000 
     2  b 3000 
     3  c 3000 
     4  d 4000 
     5  e 1000 
     6  f 1000 

агрегата возвращает data.frame и typply массив, до вас, что вы хотите, чтобы продолжить работу с результатами ,

Answer 2

ar1 <- arrange(RDD_aps,desc(RDD_aps$sale_amount)) 
collect(summarize(groupBy(ar1,ar1‌​$agent),most_expensi‌​ve=first(ar1$product‌​)))