Apache Beam Windowing on a signals phase

Question

Updated: Is it possible to window a data stream on a signals phase.

For example, there is a stream of timestamp, key, value:

[<t0, k1, 0>, <t1, k1, 98>, <t2, k1, 145>, <t4, k1, 0>, <t3, k1, 350>, <t5, k1, 40>, <t6, k1, 65>, <t7, k1, 120>, <t8, k1, 240>, <t9, k1, 352>].

The output would be two windows for key k1:

t0 - t3: [0, 98, 145, 350]
t4 - t9: [0, 40, 65, 120, 240, 352]

E.g. every time the value hits 0, start a new window for the group.

Guillem Xercavins Guillem Xercavins · Accepted Answer · 2019-06-09T09:18:41

After your question edit and use case clarification I would recommend to look into custom windowing to extend the standard sessions. As a starting point I built the following example (it can be improved upon).

Through WindowFn.AssignContext we can access the element() that it's being windowed into a proto-session. If it's equal to a given stopValue the window length will be confined to the minimum instead of using gapDuration for that purpose:

@Override
public Collection<IntervalWindow> assignWindows(AssignContext c) {
  Duration newGap = c.element().getValue().equals(this.stopValue) ? new Duration(1) : gapDuration;
  return Arrays.asList(new IntervalWindow(c.timestamp(), newGap));
}

Then, when merging the sorted windows we'll check if they do overlap but also that the window duration is not equal to 1 ms.

Collections.sort(sortedWindows);
List<MergeCandidate> merges = new ArrayList<>();
MergeCandidate current = new MergeCandidate();
for (IntervalWindow window : sortedWindows) {
  // get window duration and check if it's a stop session request
  Long windowDuration = new Duration(window.start(), window.end()).getMillis();

  if (current.intersects(window) && !windowDuration.equals(1L)) {
    current.add(window);
  } else {
    merges.add(current);
    current = new MergeCandidate(window);
  }
}
merges.add(current);
for (MergeCandidate merge : merges) {
  merge.apply(c);
}

Of course, we also can add some code so that we can provide different stopping values: a stopValue field, a withStopValue method, constructors, display data if using the Dataflow Runner, etc.

/** Value that closes the session. */
private final Integer stopValue;

/** Creates a {@code StopSessions} {@link WindowFn} with the specified gap duration. */
public static StopSessions withGapDuration(Duration gapDuration) {
  return new StopSessions(gapDuration, 0);
}

/** Creates a {@code StopSessions} {@link WindowFn} with the specified stop value. */
public StopSessions withStopValue(Integer stopValue) {
  return new StopSessions(gapDuration, stopValue);
}

/** Creates a {@code StopSessions} {@link WindowFn} with the specified gap duration and stop value. */
private StopSessions(Duration gapDuration, Integer stopValue) {
  this.gapDuration = gapDuration;
  this.stopValue = stopValue;

Now in our pipeline we can import and use the new StopSessions class with:

import org.apache.beam.sdk.transforms.windowing.StopSessions; // custom one
...

.apply("Window into StopSessions", Window.<KV<String, Integer>>into(StopSessions
  .withGapDuration(Duration.standardSeconds(10))
  .withStopValue(0)))

To mimic your example we create some data with:

.apply("Create data", Create.timestamped(
    TimestampedValue.of(KV.of("k1", 0), new Instant()), // <t0, k1, 0>
    TimestampedValue.of(KV.of("k1",98), new Instant().plus(1000)), // <t1, k1, 98>
    TimestampedValue.of(KV.of("k1",145), new Instant().plus(2000)), // <t2, k1, 145>
    TimestampedValue.of(KV.of("k1",0), new Instant().plus(4000)), // <t4, k1, 0>
    ...

With standard sessions the output would be:

user=k1, scores=[0,145,350,120,0,40,65,98,240,352], window=[2019-06-08T19:13:46.785Z..2019-06-08T19:14:05.797Z)

And with the custom one I get the following:

user=k1, scores=[350,145,98], window=[2019-06-08T21:18:51.395Z..2019-06-08T21:19:03.407Z)
user=k1, scores=[0], window=[2019-06-08T21:18:54.407Z..2019-06-08T21:18:54.408Z)
user=k1, scores=[65,240,352,120,40], window=[2019-06-08T21:18:55.407Z..2019-06-08T21:19:09.407Z)
user=k1, scores=[0], window=[2019-06-08T21:18:50.395Z..2019-06-08T21:18:50.396Z)

Changing the stopValue with .withStopValue(<int>) works as expected. The 98, 145 and 350 events are in a different session than the rest. Please note that this is not exactly like in the description as the stopValue gets assigned to a separate window instead of the new one but it can be filtered downstream and it gives you an idea on how to proceed. I would like to revisit this and also look for a Python implementation, too.

All files here.

Apache Beam Windowing on a signals phase

2 Answers