From b1a1f9b151443889ced60259ad544596bdfa5bbd Mon Sep 17 00:00:00 2001
From: "Novotnik, Petr" <petr.novotnik@firma.seznam.cz>
Date: Tue, 21 Mar 2017 09:37:45 +0100
Subject: [PATCH] #! [euphoria-flink] Avoid extra shuffle when windowing on
 streaming

---
 .../euphoria/flink/streaming/ReduceStateByKeyTranslator.java  | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/ReduceStateByKeyTranslator.java b/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/ReduceStateByKeyTranslator.java
index 978b6cc3..a3711931 100644
--- a/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/ReduceStateByKeyTranslator.java
+++ b/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/ReduceStateByKeyTranslator.java
@@ -73,7 +73,9 @@ public DataStream<?> translate(FlinkOperator<ReduceStateByKey> operator,
     // assign windows
     DataStream<KeyedMultiWindowedElement> windowed = input.map(new WindowAssigner(
             windowing, keyExtractor, valueExtractor, eventTimeAssigner))
-            .setParallelism(operator.getParallelism());
+            // ~ execute in the same chain of the input's processing
+            // so far, thereby, avoiding an unnecessary shuffle
+            .setParallelism(input.getParallelism());
 
     DataStream<WindowedElement<?, Pair>> reduced = (DataStream) windowed.keyBy(new KeyExtractor())
             .transform(operator.getName(), TypeInformation.of(WindowedElement.class), new WindowOperator<>(