[NEMO-267] Consider watermark holds in GroupByKeyAndWindowDoFnTransform #151

taegeonum · 2018-11-06T06:55:43Z

JIRA: NEMO-267: Consider watermark holds in GroupByKeyAndWindowDoFnTransform

Major changes:

Fix GroupByKeyAndWindowDoFnTransform to emit output watermarks by holding watermarks for each key

Tests for the changes:

Fix GroupByKeyAndWindowDoFnTransform by using sliding windows

…o Nemo-267

johnyangk

Thanks @taegeonum. I've left comments.

johnyangk · 2018-11-07T00:18:20Z

.../java/org/apache/nemo/compiler/frontend/beam/transform/GroupByKeyAndWindowDoFnTransform.java

@@ -122,8 +130,6 @@ public void onData(final WindowedValue<KV<K, InputT>> element) {
  private void processElementsAndTriggerTimers(final Watermark inputWatermark,


handleWatermark()?

since we do all of the following three
(1) processElements
(2) TriggerTimers
(3) emits a watermark (if necessary) to downstream

I will separate the emitOuptutWatermark and keep this method name.

johnyangk · 2018-11-07T00:34:10Z

.../java/org/apache/nemo/compiler/frontend/beam/transform/GroupByKeyAndWindowDoFnTransform.java

+   * Output watermark
+   * = max(prev output watermark,
+   *          min(input watermark, watermark holds)).
+   * @param inputWatermark input watermark


Can you add a brief note on the guarantees about the input watermark?

Something like:
{@ link InputWatermarkManager} guarantees the input watermark here to be the minimum value of the watermarks we've seen from all of the input streams.

This is already written in Transform.onWatermark :)

johnyangk · 2018-11-07T00:47:39Z

.../java/org/apache/nemo/compiler/frontend/beam/transform/GroupByKeyAndWindowDoFnTransform.java

+    }
+
+    // Find min watermark hold
+    final Watermark watermarkHold = Collections.min(keyAndWatermarkHoldMap.values());


minOfWatermarkHolds

johnyangk · 2018-11-07T00:55:31Z

.../java/org/apache/nemo/compiler/frontend/beam/transform/GroupByKeyAndWindowDoFnTransform.java

+        + "inputWatermark: {}, outputWatermark: {}", watermarkHold, inputWatermark, prevOutputWatermark);
+    }
+
+    if (watermarkHold.getTimestamp() > inputWatermark.getTimestamp()) {


Use Math.max and Math.min?

newOutputWatermark = Math.max(prevOUtputWatermark (Math.min (inputWatermark, minOfWatermarkHolds))

if (newOutputWatermark > prevOutputWatermark) {
// progress

if (minOfWatermarkHolds <= inputWatermark ) {
// Remove minimum watermark holds
}
}

johnyangk · 2018-11-07T01:11:45Z

.../java/org/apache/nemo/compiler/frontend/beam/transform/GroupByKeyAndWindowDoFnTransform.java

+   * @param inputWatermark input watermark
+   */
+  private void emitOutputWatermark(final Watermark inputWatermark) {
+


For completeness, can you check all of the three watermarks that make up the output watermark?

// prev output watermark - No need to check (?)
// inputWatermark - check for Long.MIN_VALUE
// keyAndWatermarkHoldMap - check emptiness

Why do we check Long.MIN_VALUE for inputWatermark?

johnyangk · 2018-11-07T01:15:51Z

.../java/org/apache/nemo/compiler/frontend/beam/transform/GroupByKeyAndWindowDoFnTransform.java

+      // adds the output timestamp to the watermark hold of each key
+      // +1 to the output timestamp because if the window is [0-5000), the timestamp is 4999
+      keyAndWatermarkHoldMap.put(output.getValue().getKey(),
+        new Watermark(output.getTimestamp().getMillis() + 1));


Is there a reference to this? (e.g., another Beam runner doing also the same)

In the case of following sequence of elements/watermark with (timestamp):

e1(1), e2(2), w1(3), e3(3)

With the +1 we see it as

e1(2=1+1), e2(3=2+1), w1(3), e3(4=3+1)

Would that result in an incorrect reordering of e2 and w1 (in the time-scale)?

The output here is windowed values, not all elements. Windowed value has its own timestamp value. This is different from the element timestamp.

Can you add a TODO here to check for different triggers?
https://issues.apache.org/jira/browse/NEMO-270

In the following case

W-4 / E-3 / E-2 / E-1 / E-1 --> GBKW

All elements are in the window of [1~4)

(1) If triggered at watermark W-4 this works perfectly
(2) If triggered early at E-1, then the downstream operators will consider the following E-1 late.

NEMO-270 can check for the correctness of (2)

johnyangk · 2018-11-07T01:20:35Z

.../java/org/apache/nemo/compiler/frontend/beam/transform/GroupByKeyAndWindowDoFnTransform.java

+    @Override
+    public void emit(final WindowedValue<KV<K, Iterable<InputT>>> output) {
+      // adds the output timestamp to the watermark hold of each key
+      // +1 to the output timestamp because if the window is [0-5000), the timestamp is 4999


For an element to belong to a window [0~5000) then its timestamp must be one of 0, 1, ... , 4999.
If you do +1 to an element with timestamp 4999, would that make the element belong to a different window [5000, 10000)?
Please let me know if I'm missing something here.

Window [0-5000) is triggered at watermark 5000. This means that if we receive [0~5000) window as an output, the watermark is >= 5000. But the window output timestamp is 4999, so I added + 1 here

johnyangk · 2018-11-07T01:22:14Z

...a/org/apache/nemo/compiler/frontend/beam/transform/GroupByKeyAndWindowDoFnTransformTest.java


    assertEquals(expectedValue, resultValue);
  }


-  // [---- window1 --------]         [--------------- window2 ---------------]
-  // ts1 -- ts2 -- ts3 -- watermark -- ts4 -- watermark2 -- ts5 --ts6 --ts7 -- watermark7


Can we keep the existing test (fixed window), and introduce the new test (sliding window) as a separate one?

Fixed window is a special case of sliding window, so I just added the sliding window test to test GBKW.

johnyangk · 2018-11-07T01:23:47Z

...a/org/apache/nemo/compiler/frontend/beam/transform/GroupByKeyAndWindowDoFnTransformTest.java

+    List<IntervalWindow> sortedWindows = new ArrayList<>(slidingWindows.assignWindows(ts1));
+    Collections.sort(sortedWindows, new Comparator<IntervalWindow>() {
+      @Override
+      public int compare(IntervalWindow o1, IntervalWindow o2) {


Use the existing IntervalWindow#compareTo?

johnyangk · 2018-11-07T01:44:06Z

...a/org/apache/nemo/compiler/frontend/beam/transform/GroupByKeyAndWindowDoFnTransformTest.java

    // windowed result for key 1
-    assertEquals(Arrays.asList(fixedwindows.assignWindow(ts4)), oc.outputs.get(0).getWindows());
-    checkOutput(KV.of("1", Arrays.asList("a")), oc.outputs.get(0).getValue());
+    assertEquals(Arrays.asList(window1), oc.outputs.get(0).getWindows());


Can you also check for the PaneInfo of the output elements?
(EARLY, ON_TIME, and LATE).

It'd be definitely nice to experiment with different triggers in the WindowingStrategy to ensure correct behaviors of the watermark holds, which I suppose allows the handling of EARLY and LATE elements, in addition to the timers mostly handling ON_TIME.

I will add TODO for the test

taegeonum · 2018-11-07T03:24:31Z

@johnyangk Thanks for the review! I've addressed your comments.

johnyangk

@taegeonum I've left one minor comment. Thanks!

johnyangk · 2018-11-07T03:47:30Z

.../java/org/apache/nemo/compiler/frontend/beam/transform/GroupByKeyAndWindowDoFnTransform.java

+      // adds the output timestamp to the watermark hold of each key
+      // +1 to the output timestamp because if the window is [0-5000), the timestamp is 4999
+      keyAndWatermarkHoldMap.put(output.getValue().getKey(),
+        new Watermark(output.getTimestamp().getMillis() + 1));


Can you add a TODO here to check for different triggers?
https://issues.apache.org/jira/browse/NEMO-270

In the following case

W-4 / E-3 / E-2 / E-1 / E-1 --> GBKW

All elements are in the window of [1~4)

(1) If triggered at watermark W-4 this works perfectly
(2) If triggered early at E-1, then the downstream operators will consider the following E-1 late.

NEMO-270 can check for the correctness of (2)

…o Nemo-267

taegeonum added 2 commits November 6, 2018 15:49

nemo-267

9ef3997

Merge branch 'master' of https://github.com/apache/incubator-nemo int…

5396a73

…o Nemo-267

taegeonum requested a review from johnyangk November 6, 2018 06:58

Merge branch 'master' of https://github.com/apache/incubator-nemo int…

3fdb6a0

…o Nemo-267

johnyangk suggested changes Nov 7, 2018

View reviewed changes

Address comments

494f103

johnyangk reviewed Nov 7, 2018

View reviewed changes

taegeonum added 2 commits November 7, 2018 13:51

add commentsg

1b6292c

Merge branch 'master' of https://github.com/apache/incubator-nemo int…

922d4a4

…o Nemo-267

johnyangk approved these changes Nov 7, 2018

View reviewed changes

johnyangk merged commit b1dad52 into apache:master Nov 7, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[NEMO-267] Consider watermark holds in GroupByKeyAndWindowDoFnTransform #151

[NEMO-267] Consider watermark holds in GroupByKeyAndWindowDoFnTransform #151

taegeonum commented Nov 6, 2018

johnyangk left a comment

johnyangk Nov 7, 2018

taegeonum Nov 7, 2018

johnyangk Nov 7, 2018

taegeonum Nov 7, 2018

johnyangk Nov 7, 2018

johnyangk Nov 7, 2018

johnyangk Nov 7, 2018

taegeonum Nov 7, 2018

johnyangk Nov 7, 2018

taegeonum Nov 7, 2018 •

edited

johnyangk Nov 7, 2018

johnyangk Nov 7, 2018

taegeonum Nov 7, 2018 •

edited

johnyangk Nov 7, 2018

taegeonum Nov 7, 2018

johnyangk Nov 7, 2018

johnyangk Nov 7, 2018

taegeonum Nov 7, 2018

taegeonum commented Nov 7, 2018

johnyangk left a comment

johnyangk Nov 7, 2018

		@@ -122,8 +130,6 @@ public void onData(final WindowedValue<KV<K, InputT>> element) {
		private void processElementsAndTriggerTimers(final Watermark inputWatermark,

[NEMO-267] Consider watermark holds in GroupByKeyAndWindowDoFnTransform #151

[NEMO-267] Consider watermark holds in GroupByKeyAndWindowDoFnTransform #151

Conversation

taegeonum commented Nov 6, 2018

johnyangk left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

taegeonum Nov 7, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

taegeonum Nov 7, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

taegeonum commented Nov 7, 2018

johnyangk left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

taegeonum Nov 7, 2018 •

edited

taegeonum Nov 7, 2018 •

edited