title: (Py) Spark - Group by user over a window of time
body: <p>I'm processing lots of log files and I'd like to move the job to Spark, but I can't figure out how to aggregate events over an event-based time window the way I can easily in Pandas.</p>
  
  <p>Here's exactly what I want to do:</p>
  
  <p>For a log file (simulated below) of users who have experienced some event, I'd like to go back in time, seven days, and return aggregates for all other columns.</p>
  
  <p>Here it is in Pandas. Any ideas how to port this to PySpark?</p>
  
  <pre><code>import pandas as pd
  df = pd.DataFrame({'user_id':[1,1,1,2,2,2], 'event':[0,1,0,0,0,1], 'other':[12, 20, 16, 84, 11, 15] , 'event_date':['2015-01-01 00:02:43', '2015-01-04 00:02:03', '2015-01-10 00:12:26', '2015-01-01 00:02:43', '2015-01-06 00:02:43', '2015-01-012 18:10:09']})
  df['event_date'] = pd.to_datetime(df['event_date'])
  df
  </code></pre>
  
  <p>Gives:</p>
  
  <pre><code>    event  event_date           other  user_id
  0   0      2015-01-01 00:02:43  12     1
  1   1      2015-01-04 00:02:03  20     1
  2   0      2015-01-10 00:12:26  16     1
  3   0      2015-01-01 00:02:43  84     2
  4   0      2015-01-06 00:02:43  11     2
  5   1      2015-01-12 18:10:09  15     2
  </code></pre>
  
  <p>I'd like to group this DataFrame by user_id, then exclude any row from aggregation where the row is older than seven days from the "event".</p>
  
  <p>In Pandas, like so:</p>
  
  <pre><code>def f(x):
      # Find event
      win = x.event == 1
  
      # Get the date when event === 1
      event_date = list(x[win]['event_date'])[0]
  
      # Construct the window
      min_date = event_date - pd.DateOffset(days=7) 
  
      # Set x to this specific date window
      x = x[(x.event_date &gt; min_date) &amp; (x.event_date &lt;= event_date)]
  
      # Aggregate other
      x['other'] = x.other.sum()
  
      return x[win] #, x[z]])
  
  
  df.groupby(by='user_id').apply(f).reset_index(drop=True)
  </code></pre>
  
  <p>Giving the desired output(one row per user, where event_date corresponds to event==1):</p>
  
  <pre><code>    event   event_date          other   user_id
  0   1       2015-01-04 00:02:03 32      1
  1   1       2015-01-12 18:10:09 26      2
  </code></pre>
  
  <p>Anyone know where to start getting this result in Spark?</p>