apache#13 [euphoria-flink] Avoid hash code collisions in batch Reduce…

…ByKeyTranslator
dmvk · May 15, 2018 · bec5c98 · bec5c98
1 parent f2d841f
commit bec5c98
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 8 deletions.
diff --git a/...ia/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/batch/ReduceByKeyTranslator.java b/...ia/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/batch/ReduceByKeyTranslator.java
@@ -184,14 +184,19 @@ public void reduce(Iterable<BatchElement<Window, Pair>> values, Collector<BatchE
     private void doReduce(Iterable<BatchElement<Window, Pair>> values,
                        org.apache.flink.util.Collector<BatchElement<Window, Pair>> out) {
 
-      // Tuple2[Window, Key] => Reduced Value
+      // Tuple2[Key, Window] => Reduced Value
       Map<Tuple2, TimestampedElement> reducedValues = new HashMap<>();
 
       for (BatchElement<Window, Pair> batchElement : values) {
         Object key = batchElement.getElement().getFirst();
         Window window = batchElement.getWindow();
 
-        Tuple2 kw = new Tuple2<>(window, key);
+        // Order of items in this Tuple2 is reversed compared to the
+        // order in KeySelector. This is made on purpose because
+        // all values in this reducer have the same hash code.
+        // Reversing the order will lead to better performance
+        // when kw is put into hash map.
+        Tuple2 kw = new Tuple2<>(key, window);
 
         // TimestampedElement holds only timestamp and reduced value.
         // Key and window is stored separately in key part of the HashMap.
@@ -215,14 +220,16 @@ private void doReduce(Iterable<BatchElement<Window, Pair>> values,
       }
 
       for (Map.Entry<Tuple2, TimestampedElement> e : reducedValues.entrySet()) {
-        Window window = (Window) e.getKey().f0;
-        Object key = e.getKey().f1;
+        Object key = e.getKey().f0;
+        Window window = (Window) e.getKey().f1;
 
-        out.collect(new BatchElement<>(
+        @SuppressWarnings("unchecked")
+        BatchElement<Window, Pair> batchElement = new BatchElement<>(
                 window,
                 e.getValue().getTimestamp(),
-                Pair.of(key, e.getValue().getElement())
-        ));
+                Pair.of(key, e.getValue().getElement()));
+
+        out.collect(batchElement);
       }
     }
 

diff --git a/...phoria-flink/src/main/java/cz/seznam/euphoria/flink/batch/ReduceStateByKeyTranslator.java b/...phoria-flink/src/main/java/cz/seznam/euphoria/flink/batch/ReduceStateByKeyTranslator.java
@@ -161,7 +161,7 @@ public void reduce(Iterable<BatchElement<?, Pair>> values,
     {
       activeReducers = new HashMap<>();
       for (BatchElement<?, Pair> batchElement : values) {
-        Object key = batchElement.getElement().getKey();
+        Object key = batchElement.getElement().getFirst();
 
         GroupReducer reducer = activeReducers.get(key);
         if (reducer == null) {