Apache Beam - Sliding Windows outputting multiple windows

Question

I'm taking a PCollection of sessions and trying to get average session duration per channel/connection. I'm doing something where my early triggers are firing for each window produced - if 60min windows sliding every 1 minute, an early trigger will fire 60 times. Looking at the timestamps on the outputs, there's a window every minute for 60minutes into the future. I'd like the trigger to fire once for the most recent window so that every 10 seconds I have an average of session durations for the last 60 minutes.

I've used sliding windows before and had the results I expected. By mixing sliding and sessions windows, I'm somehow causing this.

Let me paint you a picture of my pipeline:

First, I'm creating sessions based on active users:

  .apply("Add Window Sessions", 
Window.<KV<String, String>> into(Sessions.withGapDuration(Duration.standardMinutes(60)))
  .withOnTimeBehavior(Window.OnTimeBehavior.FIRE_ALWAYS)
  .triggering(
   AfterWatermark.pastEndOfWindow()
        .withEarlyFirings(AfterProcessingTime
        .pastFirstElementInPane()
        .plusDelayOf(Duration.standardSeconds(10))))
  .withAllowedLateness(Duration.ZERO)
  .discardingFiredPanes()
 )
 .apply("Group Sessions", Latest.perKey())

Steps after this create a session object, compute session duration, etc. This ends with a PCollection(Session).

I create a KV of connection,duration from the Pcollection(Session).

Then I apply the sliding window and then the mean.

    .apply("Apply Rolling Minute Window",
      Window. < KV < String, Integer >> into(
       SlidingWindows
       .of(Duration.standardMinutes(60))
       .every(Duration.standardMinutes(1)))
      .triggering(
       Repeatedly.forever(
        AfterWatermark.pastEndOfWindow()
        .withEarlyFirings(AfterProcessingTime
         .pastFirstElementInPane()
         .plusDelayOf(Duration.standardSeconds(10)))
       )
      )
      .withAllowedLateness(Duration.standardMinutes(1))
      .discardingFiredPanes()
     )
     .apply("Get Average", Mean.perKey())

It's at this point where I'm seeing issues. What I'd like to see is a single output per key with the average duration. What I'm actually seeing is 60 outputs for the same key for each minute into the next 60 minutes.

With this log in a DoFn with C being the ProcessContext:

LOG.info(c.pane().getTiming() + " " + c.timestamp());

I get this output 60 times with timestamps 60 minutes into the future:

EARLY 2017-12-17T20:41:59.999Z
EARLY 2017-12-17T20:43:59.999Z
EARLY 2017-12-17T20:56:59.999Z
(cont)

The log was printed at Dec 17, 2017 19:35:19. The number of outputs is always window size/slide duration. So if I did 60 minute windows every 5 minutes, I would get 12 output.

bmark bmark · Accepted Answer · 2017-12-18T02:26:04

I think I've made sense of this.

Sliding windows create a new window with the .every() function. Setting early firings applies to each window so getting multiple firings makes sense.

In order to fit my use case and only output the "current window", I'm checking c.pane().isFirst() == true before outputting results and adjusting the .every() to control the frequency.

Apache Beam - Sliding Windows outputting multiple windows

1 Answers