#27 [euphoria-flink] Rewrite windowing to native implementation of StreamOperator #50

vanekjar · 2017-03-19T11:21:26Z

Finally, my PR is ready.

The code for handling windows is still over-complicated, but I deleted more code than actually added which is always a good sign.

Instead of using ProcessFunction I directly implemented custom StreamOperator. It turned out during implementation that ProcessFunction API is not so powerful for our use case - for example it's not possible to access the current key in callbacks.

There are a few pitfalls in the current PR that I am not very satisfied with:

Rewriting windowing haven't brought any performance benefit. Runtime is more or less the same as with previous implementation. On the other hand it now opens new opportunities to optimize - for example removing duplicate timestamp from WindowedElement flowing through the pipeline and use the built-in Flink timestamp instead. This will come in another PR soon.
I had to remove the functionality of flushing remaining windows in case of bounded stream end. So far all remaining windows were flushed to output in case of EOS. It proved that tracking of all existing windows had a huge performance drawback considering the registered window set must be persisted for fault-tolerance. I left that functionality only for unit testing (works for TestFlinkExecutor in local mode). Now it opens a discussion if we need this functionality of bounded stream in production.

Thanks for the review.

Resolves: euphoria-flink: upgrade to flink 1.2.x #33
Closes: Flink executor: Make use of ProcessFunction #27

…reamOperator

xitep · 2017-03-20T10:39:15Z

euphoria-operator-testkit/src/main/java/cz/seznam/euphoria/operator/test/ReduceByKeyTest.java


        ArrayList<Long> assignerTimes = new ArrayList<>(TETETS_SEEN_TIMES_ASSIGNER);
        assignerTimes.sort(Comparator.naturalOrder());
-        assertEquals(asList(15_000L, 19_999L, 25_000L, 29_999L), assignerTimes);
+        assertEquals(asList(19_999L, 19_999L, 29_999L, 29_999L), assignerTimes);


not sure we need this test anymore. i think testElementTimestamp should suffice.

xitep · 2017-03-20T10:42:11Z

...rc/test/java/cz/seznam/euphoria/flink/streaming/windowing/KeyedMultiWindowedElementTest.java

+                    .createSerializer(new ExecutionConfig());
+
+    // must be POJO serializer for performance reasons
+    assertTrue(serializer instanceof PojoSerializer);


thanks! 👍

xitep · 2017-03-20T10:57:02Z

euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/ReduceByKeyTranslator.java

-import java.util.Iterator;
-import java.util.Set;
-
-class ReduceByKeyTranslator implements StreamingOperatorTranslator<ReduceByKey> {


no specialization for the ReduceByKey operator in the streaming executor anymore?

It's easier to maintain just one implementation. Flink is internally doing the same job with incremental reducer as we do now in WindowOperator.
Anyway the performance of benchmark using ReduceByKey is the same with the new implementation.

i agree it's easier to maintain only one implementation. no doubt about that. what i worry about a bit is the difference between serializing the state (into the backend storage) of a combining RBK (a single value) vs. serializing the same value in a list (of size one.) my guess is, we cannot tell this difference now, since other factors are likely to hide this overhead. let's stick with one implementation and optimize later - if necessary.

I agree with your concern about a difference between ValueStorage and ListStorage. That can really make a difference.
There is a chance to optimize the internals of ReduceByKey.ReduceState in a way that it will use ValueStorage in case it's combinable. Also the storage is accessed (serialized) multiple times now when adding to state, this can be also avoided.

if necessary, i think we can even optimize this later directly in the api layer in #getBasicOps of the RBK operator. this would allow us to specialize the case without having to maintain separate impls in the executors.

xitep · 2017-03-20T11:05:43Z

euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/StreamingWindowedElement.java

-  public StreamingWindowedElement(W window, long timestamp, T element) {
-    super(window, timestamp, element);
-  }
-}


many thanks for the clean-up!

…mpEarlyTriggeredStreaming)

xitep · 2017-03-20T12:13:24Z

euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/ReduceStateByKeyTranslator.java

+    DataStream<WindowedElement<?, Pair>> reduced = (DataStream) windowed.keyBy(new KeyExtractor())
+            .transform(operator.getName(), TypeInformation.of(WindowedElement.class), new WindowOperator<>(
+                    windowing, stateFactory, stateCombiner, context.isLocalMode()))
+            .setParallelism(operator.getParallelism());


i would like to see support for running this "as is", as well as with the value extraction and window assignment functionality executing only after the shuffle (maybe some global parameter to the translation layer). at least for our benchmarking this will be necessary. what do you think it would take to support both?

Definitely agree it would be useful. But I am not sure if this is not the part of #47

oh yeah, right. let's introduce that with the mentioned ticket later.

xitep · 2017-03-20T12:37:40Z

euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/windowing/WindowOperator.java

+                  List<State> states = new ArrayList<>();
+                  states.add(getWindowState(stateResultWindow));
+                  mergedStateWindows.forEach(sw -> states.add(getWindowState(sw)));
+                  stateCombiner.apply(states);


in regards to the above FIXME, i wanted to suggest to change the type of the state-combiner. however, now i see it's a CombinableReduceFunction which does have a return value that is supposed to replace the merged states. in the inmem as well as in the spark executor (e.g. GroupedReducer) we are doing so. if i'm not mistaken applying the same technique here should resolve the above FIXME.

I am not sure about that. The returned state from CombinableReduceFunction doesn't matter at all. Since our state is basically "stateless", it depends if the resulting state stored the merged value to the appropriate persistent storage using correct namespace.
In this case we need the state combiner to store the result in stateResultWindow namespace. And I can't see any method how to ensure that.

ah, thanks for the explanation! now i see it. we'll follow up on this later - combinable-reduce-function then doesn't seem right to me at this place. may i ask you to set up a ticket for that FIXME?

Issue created #51

xitep · 2017-03-20T12:50:28Z

euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/windowing/MergingWindowSet.java

+ */
+class MergingWindowSet<W extends Window> {
+
+  private final MergingWindowing windowing;


this is quite cool 😎! do you think we can re-use it to replace parts of GroupedReducer in euphoria-core?

ah, github is playing tricks on me ;) ... that comment was meant to address the class as a whole.

I don't think reusing this class would bring any benefit. It's too complicated because the window set must be persisted after each step in streaming. Most of the code is performance optimization to avoid costly persistent state allocation. This is not the case in GroupReducer where everything is processed in a plain HashMap in memory.

xitep · 2017-03-20T12:59:59Z

...oria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/windowing/AttachedWindowing.java

+    public TriggerResult onElement(long time, WID window, TriggerContext ctx) {
+      // FIXME batch window shouldn't be used in stream flow in the future
+      // issue #38 on GitHub
+      if (window instanceof Batch.BatchWindow) return TriggerResult.NOOP;


what about throwing an exception here?

Unfortunately there still exists unit test with unbounded source and no windowing. I think it needs to be resolved with the issue #38

ok, fine with me.

xitep · 2017-03-20T13:03:26Z

euphoria-flink/src/main/java/cz/seznam/euphoria/flink/ExecutionEnvironment.java

@@ -22,6 +22,7 @@
 import cz.seznam.euphoria.core.client.dataset.windowing.WindowedElement;
 import cz.seznam.euphoria.core.client.flow.Flow;
 import cz.seznam.euphoria.core.client.util.Pair;
+import cz.seznam.euphoria.flink.streaming.windowing.KeyedMultiWindowedElement;


looks like an unused import

xitep · 2017-03-20T13:23:40Z

euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/ReduceStateByKeyTranslator.java

+            windowing, keyExtractor, valueExtractor, eventTimeAssigner))
+            .setParallelism(operator.getParallelism());
+
+    DataStream<WindowedElement<?, Pair>> reduced = (DataStream) windowed.keyBy(new KeyExtractor())


i think we're better off with .keyBy("key"); flink will then derive the type information from the input data stream automatically; it'll still be "object" at this moment since we don't supply enough type information through the WindowAssigner, but that's about to come in some future.

Great idea, but unfortunately doesn't work. Or at least I don't know how to make it work. Object is not a key type.

org.apache.flink.api.common.InvalidProgramException: This type (GenericType<java.lang.Object>) cannot be used as key.

hm ... we'll need more type information :/ anyway, thanks for having a try!

xitep · 2017-03-20T14:18:03Z

euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/windowing/WindowOperator.java

+      windowState = getWindowState(stateWindow);
+    } else {
+      windowState = getWindowState(window);
+    }


potential micro-optimization possibility here: don't lookup out the window state from the state table when not necessary (e.g. tr == NOOP)

xitep · 2017-03-20T14:52:07Z

thank you so much for cleaning up old stuff! feel free to merge into master. it looks very good to me.

the fact that the performance didn't get better means we didn't get worse! :) (our benchmark is just a single scenario and it happens to hit a particular bottleneck which is present in both versions.) it be interesting to compare them without the bottleneck eliminated.

#27 [euphoria-flink] Rewrite windowing to native implementation of St…

5e916a8

…reamOperator

vanekjar assigned xitep Mar 19, 2017

xitep reviewed Mar 20, 2017

View reviewed changes

#27 Remove useless unit test for early triggering (testElementTimesta…

c887c8e

…mpEarlyTriggeredStreaming)

xitep reviewed Mar 20, 2017

View reviewed changes

#27 Micro-optimization in case when trigger result is NOOP

b368ce0

vanekjar merged commit 3f67275 into master Mar 20, 2017

vanekjar deleted the 27/ProcessFunction branch March 20, 2017 16:04

xitep mentioned this pull request Mar 21, 2017

#! [euphoria-flink] Avoid extra shuffle when windowing on streaming #52

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

#27 [euphoria-flink] Rewrite windowing to native implementation of StreamOperator #50

#27 [euphoria-flink] Rewrite windowing to native implementation of StreamOperator #50

vanekjar commented Mar 19, 2017 •

edited by xitep

Loading

xitep Mar 20, 2017

xitep Mar 20, 2017

xitep Mar 20, 2017

vanekjar Mar 20, 2017

xitep Mar 20, 2017

vanekjar Mar 20, 2017 •

edited

Loading

xitep Mar 20, 2017

xitep Mar 20, 2017

xitep Mar 20, 2017

vanekjar Mar 20, 2017

xitep Mar 20, 2017

xitep Mar 20, 2017 •

edited

Loading

vanekjar Mar 20, 2017

xitep Mar 20, 2017

vanekjar Mar 20, 2017

xitep Mar 20, 2017

xitep Mar 20, 2017

vanekjar Mar 20, 2017

xitep Mar 20, 2017

vanekjar Mar 20, 2017

xitep Mar 20, 2017

xitep Mar 20, 2017

xitep Mar 20, 2017

vanekjar Mar 20, 2017 •

edited

Loading

xitep Mar 20, 2017

xitep Mar 20, 2017

xitep commented Mar 20, 2017

#27 [euphoria-flink] Rewrite windowing to native implementation of StreamOperator #50

#27 [euphoria-flink] Rewrite windowing to native implementation of StreamOperator #50

Conversation

vanekjar commented Mar 19, 2017 • edited by xitep Loading

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

vanekjar Mar 20, 2017 • edited Loading

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

xitep Mar 20, 2017 • edited Loading

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vanekjar Mar 20, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xitep commented Mar 20, 2017

vanekjar commented Mar 19, 2017 •

edited by xitep

Loading

vanekjar Mar 20, 2017 •

edited

Loading

xitep Mar 20, 2017 •

edited

Loading

vanekjar Mar 20, 2017 •

edited

Loading