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

seznam · Mar 21, 2017 · 4312951 · 4312951
1 parent 3f67275
commit 4312951
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/...ia-flink/src/main/java/cz/seznam/euphoria/flink/streaming/ReduceStateByKeyTranslator.java b/...ia-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<>(