r语言 - SparkR中用户自定义的聚合函数

Question

我有这样的邮件记录:

Name MailingID  Timestamp    Event
1 John         1 2014-04-18     Sent
2 John         2 2015-04-21     Sent
3 Mary         1 2015-04-22 Returned
4 Mary         2 2015-04-25     Sent
5 John         1 2015-05-01  Replied

，可以创建为DataFrame:

df <- createDataFrame(sqlContext, data.frame(Name = c('John','John','Mary','Mary','John'),
                                             MailingID = c(1,2,1,2,1),
                                             Timestamp=c('2014-04-18','2015-04-21','2015-04-22','2015-04-25','2015-05-01'),
                                             Event=c('Sent','Sent','Returned','Sent','Replied')))

我想知道谁回复了最近发给他/她的2封邮件中的任何一封，所以使用摘要帮助函数和dplyr我可以做:

localDf <- collect(df)
library(lubridate)
library(magrittr)
library(dplyr)
hasRepliedLatest <- function(MailingID, Timestamp, Event, Latest_N) {
  length(intersect(MailingID[Event == 'Replied'], MailingID[Event == 'Sent'][1:Latest_N])) > 0
}
localDf %>%
  arrange(desc(Timestamp)) %>%
  group_by(Name) %>%
  summarize(RepliedLatest = hasRepliedLatest(MailingID, Timestamp, Event, 2))
detach(package:dplyr) # to avoid function confliction with SparkR

结果是:

  Name RepliedLatest
1 John          TRUE
2 Mary         FALSE

现在我想在SparkR上这样做，即在DataFrame上而不是在本地data.frame上。所以我试了:

df %>%
  arrange(desc(df$Timestamp)) %>%
  group_by(df$Name) %>%
  summarize(RepliedLatest = hasRepliedLatest(df$MailingID, df$Timestamp, df$Event, 2))

然后我得到错误说我的函数不会与S4类DataFrame工作。如何在SparkR中正确地做到这一点?也欢迎使用sparkRHive.init或sparkRSQL.init创建的sqlContext的SQL查询解决方案。

Answer 1

SparkSQL <= 1.4不支持用户定义的聚合函数，据我所知SparkR根本没有udf，所以除非您使用当前的开发分支或1.5 RC udf，否则不支持udf。

我仍然不确定我是否理解你的数据模型和逻辑，但你可以尝试这样做:

# Select last 2 sent events and all other which occurred in this window
tmp <- sql(sqlContext,    
   "SELECT *, SUM(CASE WHEN event = 'Sent' THEN 1 ELSE 0 END) OVER w AS ind
    FROM df WHERE Event IN ('Sent', 'Replied')
    HAVING ind <= 2
    WINDOW w AS (PARTITION BY name ORDER BY DATE(Timestamp) DESC)")

# Split sent and replied
sent <- tmp %>% filter(tmp$Event == "Sent")
replied <- tmp %>% filter(tmp$Event == "Replied")
registerTempTable(sent,  "sent")
registerTempTable(replied,  "replied")
# Join and count
sql(sqlContext,
    "SELECT
        sent.name,
        SUM(
            CASE WHEN replied.event IS NOT NULL THEN 1
            ELSE 0 END
        ) > 0 AS repliedlatest 
     FROM sent LEFT JOIN replied ON
        sent.name = replied.name AND
        sent.mailingid = replied.mailingid
     -- Not part of the original logic
     WHERE DATE(sent.timestamp) <= DATE(replied.timestamp) 
     GROUP BY sent.name") %>% head()