title: User defined aggregate function in SparkR
body: <p>I have recods of mailings like this:</p>
  
  <pre><code>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
  </code></pre>
  
  <p>which can be created as <code>DataFrame</code>:</p>
  
  <pre><code>df &lt;- 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')))
  </code></pre>
  
  <p>I want to find out who has replied any of the 2 latest mails sent to him/her, so with a summary helper function and <code>dplyr</code> I can do:</p>
  
  <pre><code>localDf &lt;- collect(df)
  
  library(lubridate)
  library(magrittr)
  library(dplyr)
  
  hasRepliedLatest &lt;- function(MailingID, Timestamp, Event, Latest_N) {
    length(intersect(MailingID[Event == 'Replied'], MailingID[Event == 'Sent'][1:Latest_N])) &gt; 0
  }
  
  localDf %&gt;%
    arrange(desc(Timestamp)) %&gt;%
    group_by(Name) %&gt;%
    summarize(RepliedLatest = hasRepliedLatest(MailingID, Timestamp, Event, 2))
  
  detach(package:dplyr) # to avoid function confliction with SparkR
  </code></pre>
  
  <p>the outcome is:</p>
  
  <pre><code>  Name RepliedLatest
  1 John          TRUE
  2 Mary         FALSE
  </code></pre>
  
  <p>Now I want to do this with <code>SparkR</code>, i.e., on <code>DataFrame</code> instead of on local <code>data.frame</code>. So I tried:</p>
  
  <pre><code>df %&gt;%
    arrange(desc(df$Timestamp)) %&gt;%
    group_by(df$Name) %&gt;%
    summarize(RepliedLatest = hasRepliedLatest(df$MailingID, df$Timestamp, df$Event, 2))
  </code></pre>
  
  <p>Then I got error saying my function won't work with S4 class <code>DataFrame</code>. How to do this correctly in <code>SparkR</code>? Solutions using SQL query with <code>sqlContext</code> created by <code>sparkRHive.init</code> or <code>sparkRSQL.init</code> are also welcome.</p>