title: Apache Beam per-user session windows are unmerged
body: <p>We have an app that has users; each user uses our app for something like 10-40 minutes per go and I would like to count the distribution/occurrences of events happing per-such-session, based on specific events having happened (e.g. "this user converted", "this user had a problem last session", "this user had a successful last session").</p>
  
  <p>(After this I'd like to count these higher-level events per day, but that's a separate question)</p>
  
  <p>For this I've been looking into session windows; but all <a href="https://beam.apache.org/documentation/programming-guide/#setting-your-pcollections-windowing-function" rel="nofollow noreferrer">docs</a> seem geared towards global session windows, but I'd like to create them per-user (which is also a natural partitioning).</p>
  
  <p>I'm having trouble finding docs (python preferred) on how to do this. Could you point me in the right direction?</p>
  
  <p>Or in other words: <strong>How do I create per-user per-session windows that can output more structured (enriched) events?</strong></p>
  
  <h2>What I have</h2>
  
  <pre><code>class DebugPrinter(beam.DoFn):
    """Just prints the element with logging"""
    def process(self, element, window=beam.DoFn.WindowParam):
      _, x = element
      logging.info("&gt;&gt;&gt; Received %s %s with window=%s", x['jsonPayload']['value'], x['timestamp'], window)
      yield element
  
  def sum_by_event_type(user_session_events):
    logging.debug("Received %i events: %s", len(user_session_events), user_session_events)
    d = {}
    for key, group in groupby(user_session_events, lambda e: e['jsonPayload']['value']):
      d[key] = len(list(group))
    logging.info("After counting: %s", d)
    return d
  
  # ...
  
  by_user = valid \
    | 'keyed_on_user_id'      &gt;&gt; beam.Map(lambda x: (x['jsonPayload']['userId'], x))
  
  session_gap = 5 * 60 # [s]; 5 minutes
  
  user_sessions = by_user \
    | 'user_session_window'   &gt;&gt; beam.WindowInto(beam.window.Sessions(session_gap),
                                                 timestamp_combiner=beam.window.TimestampCombiner.OUTPUT_AT_EOW) \
    | 'debug_printer'         &gt;&gt; beam.ParDo(DebugPrinter()) \
    | beam.CombinePerKey(sum_by_event_type)
  </code></pre>
  
  <h2>What it outputs</h2>
  
  <pre><code>INFO:root:&gt;&gt;&gt; Received event_1 2019-03-12T08:54:29.200Z with window=[1552380869.2, 1552381169.2)
  INFO:root:&gt;&gt;&gt; Received event_2 2019-03-12T08:54:29.200Z with window=[1552380869.2, 1552381169.2)
  INFO:root:&gt;&gt;&gt; Received event_3 2019-03-12T08:54:30.400Z with window=[1552380870.4, 1552381170.4)
  INFO:root:&gt;&gt;&gt; Received event_4 2019-03-12T08:54:36.300Z with window=[1552380876.3, 1552381176.3)
  INFO:root:&gt;&gt;&gt; Received event_5 2019-03-12T08:54:38.100Z with window=[1552380878.1, 1552381178.1)
  </code></pre>
  
  <p>So as you can see; the Session() window doesn't expand the Window, but groups only very close events together... What's being done wrong?</p>