You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mb...@apache.org on 2016/03/02 17:22:43 UTC
[1/2] flink git commit: [FLINK-3422][streaming][api-breaking]
Scramble HashPartitioner hashes.
Repository: flink
Updated Branches:
refs/heads/release-1.0 23dc2a4ac -> b862fd0b3
[FLINK-3422][streaming][api-breaking] Scramble HashPartitioner hashes.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a049d80e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a049d80e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a049d80e
Branch: refs/heads/release-1.0
Commit: a049d80e8aef7f0d23fbc06d263fb3e7a0f2f05f
Parents: 23dc2a4
Author: Gabor Horvath <xa...@gmail.com>
Authored: Sun Feb 21 14:54:44 2016 +0100
Committer: Márton Balassi <mb...@apache.org>
Committed: Wed Mar 2 12:47:10 2016 +0100
----------------------------------------------------------------------
.../operators/hash/CompactingHashTable.java | 29 ++--------
.../operators/shipping/OutputEmitter.java | 31 +---------
.../apache/flink/runtime/util/MathUtils.java | 59 +++++++++++++++++++-
.../runtime/partitioner/HashPartitioner.java | 3 +-
.../apache/flink/streaming/api/IterateTest.java | 5 +-
.../api/scala/StreamingOperatorsITCase.scala | 11 +++-
6 files changed, 78 insertions(+), 60 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a049d80e/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/CompactingHashTable.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/CompactingHashTable.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/CompactingHashTable.java
index 5a0c6cc..b4d03e7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/CompactingHashTable.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/CompactingHashTable.java
@@ -332,7 +332,7 @@ public class CompactingHashTable<T> extends AbstractMutableHashTable<T> {
return;
}
- final int hashCode = hash(this.buildSideComparator.hash(record));
+ final int hashCode = MathUtils.jenkinsHash(this.buildSideComparator.hash(record));
final int posHashCode = hashCode % this.numBuckets;
// get the bucket for the given hash code
@@ -360,7 +360,7 @@ public class CompactingHashTable<T> extends AbstractMutableHashTable<T> {
return;
}
- final int searchHashCode = hash(this.buildSideComparator.hash(record));
+ final int searchHashCode = MathUtils.jenkinsHash(this.buildSideComparator.hash(record));
final int posHashCode = searchHashCode % this.numBuckets;
// get the bucket for the given hash code
@@ -1140,26 +1140,7 @@ public class CompactingHashTable<T> extends AbstractMutableHashTable<T> {
this.compactionMemory.resetRWViews();
this.compactionMemory.pushDownPages();
}
-
- /**
- * This function hashes an integer value. It is adapted from Bob Jenkins' website
- * <a href="http://www.burtleburtle.net/bob/hash/integer.html">http://www.burtleburtle.net/bob/hash/integer.html</a>.
- * The hash function has the <i>full avalanche</i> property, meaning that every bit of the value to be hashed
- * affects every bit of the hash value.
- *
- * @param code The integer to be hashed.
- * @return The hash code for the integer.
- */
- private static int hash(int code) {
- code = (code + 0x7ed55d16) + (code << 12);
- code = (code ^ 0xc761c23c) ^ (code >>> 19);
- code = (code + 0x165667b1) + (code << 5);
- code = (code + 0xd3a2646c) ^ (code << 9);
- code = (code + 0xfd7046c5) + (code << 3);
- code = (code ^ 0xb55a4f09) ^ (code >>> 16);
- return code >= 0 ? code : -(code + 1);
- }
-
+
/**
* Iterator that traverses the whole hash table once
*
@@ -1286,7 +1267,7 @@ public class CompactingHashTable<T> extends AbstractMutableHashTable<T> {
if (closed) {
return null;
}
- final int searchHashCode = hash(this.probeTypeComparator.hash(probeSideRecord));
+ final int searchHashCode = MathUtils.jenkinsHash(this.probeTypeComparator.hash(probeSideRecord));
final int posHashCode = searchHashCode % numBuckets;
@@ -1359,7 +1340,7 @@ public class CompactingHashTable<T> extends AbstractMutableHashTable<T> {
if (closed) {
return null;
}
- final int searchHashCode = hash(this.probeTypeComparator.hash(probeSideRecord));
+ final int searchHashCode = MathUtils.jenkinsHash(this.probeTypeComparator.hash(probeSideRecord));
final int posHashCode = searchHashCode % numBuckets;
http://git-wip-us.apache.org/repos/asf/flink/blob/a049d80e/flink-runtime/src/main/java/org/apache/flink/runtime/operators/shipping/OutputEmitter.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/shipping/OutputEmitter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/shipping/OutputEmitter.java
index 92e3787..fdbcd9f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/shipping/OutputEmitter.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/shipping/OutputEmitter.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.runtime.io.network.api.writer.ChannelSelector;
import org.apache.flink.runtime.plugable.SerializationDelegate;
+import org.apache.flink.runtime.util.MathUtils;
/**
* The output emitter decides to which of the possibly multiple output channels a record is sent.
@@ -187,39 +188,11 @@ public class OutputEmitter<T> implements ChannelSelector<SerializationDelegate<T
private int[] hashPartitionDefault(T record, int numberOfChannels) {
int hash = this.comparator.hash(record);
- hash = murmurHash(hash);
+ this.channels[0] = MathUtils.murmurHash(hash) % numberOfChannels;
- if (hash >= 0) {
- this.channels[0] = hash % numberOfChannels;
- }
- else if (hash != Integer.MIN_VALUE) {
- this.channels[0] = -hash % numberOfChannels;
- }
- else {
- this.channels[0] = 0;
- }
-
return this.channels;
}
- private int murmurHash(int k) {
- k *= 0xcc9e2d51;
- k = Integer.rotateLeft(k, 15);
- k *= 0x1b873593;
-
- k = Integer.rotateLeft(k, 13);
- k *= 0xe6546b64;
-
- k ^= 4;
- k ^= k >>> 16;
- k *= 0x85ebca6b;
- k ^= k >>> 13;
- k *= 0xc2b2ae35;
- k ^= k >>> 16;
-
- return k;
- }
-
private final int[] rangePartition(final T record, int numberOfChannels) {
if (this.channels == null || this.channels.length != 1) {
this.channels = new int[1];
http://git-wip-us.apache.org/repos/asf/flink/blob/a049d80e/flink-runtime/src/main/java/org/apache/flink/runtime/util/MathUtils.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/MathUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/MathUtils.java
index 939a2e1..2acc55c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/MathUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/MathUtils.java
@@ -109,7 +109,64 @@ public final class MathUtils {
public static boolean isPowerOf2(long value) {
return (value & (value - 1)) == 0;
}
-
+
+ /**
+ * This function hashes an integer value. It is adapted from Bob Jenkins' website
+ * <a href="http://www.burtleburtle.net/bob/hash/integer.html">http://www.burtleburtle.net/bob/hash/integer.html</a>.
+ * The hash function has the <i>full avalanche</i> property, meaning that every bit of the value to be hashed
+ * affects every bit of the hash value.
+ *
+ * It is crucial to use different hash functions to partition data across machines and the internal partitioning of
+ * data structures. This hash function is intended for partitioning internally in data structures.
+ *
+ * @param code The integer to be hashed.
+ * @return The non-negative hash code for the integer.
+ */
+ public static int jenkinsHash(int code) {
+ code = (code + 0x7ed55d16) + (code << 12);
+ code = (code ^ 0xc761c23c) ^ (code >>> 19);
+ code = (code + 0x165667b1) + (code << 5);
+ code = (code + 0xd3a2646c) ^ (code << 9);
+ code = (code + 0xfd7046c5) + (code << 3);
+ code = (code ^ 0xb55a4f09) ^ (code >>> 16);
+ return code >= 0 ? code : -(code + 1);
+ }
+
+ /**
+ * This function hashes an integer value.
+ *
+ * It is crucial to use different hash functions to partition data across machines and the internal partitioning of
+ * data structures. This hash function is intended for partitioning across machines.
+ *
+ * @param code The integer to be hashed.
+ * @return The non-negative hash code for the integer.
+ */
+ public static int murmurHash(int code) {
+ code *= 0xcc9e2d51;
+ code = Integer.rotateLeft(code, 15);
+ code *= 0x1b873593;
+
+ code = Integer.rotateLeft(code, 13);
+ code *= 0xe6546b64;
+
+ code ^= 4;
+ code ^= code >>> 16;
+ code *= 0x85ebca6b;
+ code ^= code >>> 13;
+ code *= 0xc2b2ae35;
+ code ^= code >>> 16;
+
+ if (code >= 0) {
+ return code;
+ }
+ else if (code != Integer.MIN_VALUE) {
+ return -code;
+ }
+ else {
+ return 0;
+ }
+ }
+
// ============================================================================================
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/a049d80e/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/partitioner/HashPartitioner.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/partitioner/HashPartitioner.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/partitioner/HashPartitioner.java
index ec0cf94..82f0141 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/partitioner/HashPartitioner.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/partitioner/HashPartitioner.java
@@ -20,6 +20,7 @@ package org.apache.flink.streaming.runtime.partitioner;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.runtime.plugable.SerializationDelegate;
+import org.apache.flink.runtime.util.MathUtils;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
/**
@@ -48,7 +49,7 @@ public class HashPartitioner<T> extends StreamPartitioner<T> {
} catch (Exception e) {
throw new RuntimeException("Could not extract key from " + record.getInstance().getValue(), e);
}
- returnArray[0] = Math.abs(key.hashCode() % numberOfOutputChannels);
+ returnArray[0] = MathUtils.murmurHash(key.hashCode()) % numberOfOutputChannels;
return returnArray;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a049d80e/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/IterateTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/IterateTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/IterateTest.java
index cd73b41..920185a 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/IterateTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/IterateTest.java
@@ -29,6 +29,7 @@ import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.util.MathUtils;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.IterativeStream;
@@ -488,9 +489,9 @@ public class IterateTest extends StreamingMultipleProgramsTestBase {
public void flatMap(Integer value, Collector<Integer> out) throws Exception {
received++;
if (key == -1) {
- key = value % 3;
+ key = MathUtils.murmurHash(value % 3) % 3;
} else {
- assertEquals(key, value % 3);
+ assertEquals(key, MathUtils.murmurHash(value % 3) % 3);
}
if (value > 0) {
out.collect(value - 1);
http://git-wip-us.apache.org/repos/asf/flink/blob/a049d80e/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/StreamingOperatorsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/StreamingOperatorsITCase.scala b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/StreamingOperatorsITCase.scala
index 60a02e7..fe49b1f 100644
--- a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/StreamingOperatorsITCase.scala
+++ b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/StreamingOperatorsITCase.scala
@@ -20,6 +20,7 @@ package org.apache.flink.streaming.api.scala
import org.apache.flink.api.common.functions.{RichMapFunction, FoldFunction}
import org.apache.flink.core.fs.FileSystem
+import org.apache.flink.runtime.util.MathUtils
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext
import org.apache.flink.test.util.TestBaseUtils
@@ -71,7 +72,7 @@ class StreamingOperatorsITCase extends ScalaStreamingMultipleProgramsTestBase {
override def run(ctx: SourceContext[(Int, Int)]): Unit = {
0 until numElements foreach {
- i => ctx.collect((i % numKeys, i))
+ i => ctx.collect((MathUtils.murmurHash(i) % numKeys, i))
}
}
@@ -86,8 +87,12 @@ class StreamingOperatorsITCase extends ScalaStreamingMultipleProgramsTestBase {
}
})
.map(new RichMapFunction[Int, (Int, Int)] {
+ var key: Int = -1
override def map(value: Int): (Int, Int) = {
- (getRuntimeContext.getIndexOfThisSubtask, value)
+ if (key == -1) {
+ key = MathUtils.murmurHash(value) % numKeys
+ }
+ (key, value)
}
})
.split{
@@ -106,7 +111,7 @@ class StreamingOperatorsITCase extends ScalaStreamingMultipleProgramsTestBase {
.javaStream
.writeAsText(resultPath2, FileSystem.WriteMode.OVERWRITE)
- val groupedSequence = 0 until numElements groupBy( _ % numKeys)
+ val groupedSequence = 0 until numElements groupBy( MathUtils.murmurHash(_) % numKeys )
expected1 = groupedSequence(0).scanLeft(0)(_ + _).tail.mkString("\n")
expected2 = groupedSequence(1).scanLeft(0)(_ + _).tail.mkString("\n")
[2/2] flink git commit: [FLINK-3422][streaming] Update tests reliant
on hashing
Posted by mb...@apache.org.
[FLINK-3422][streaming] Update tests reliant on hashing
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b862fd0b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b862fd0b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b862fd0b
Branch: refs/heads/release-1.0
Commit: b862fd0b3657d8b9026a54782bad5a1fb71c19f4
Parents: a049d80
Author: Márton Balassi <mb...@apache.org>
Authored: Sun Feb 21 23:01:00 2016 +0100
Committer: Márton Balassi <mb...@apache.org>
Committed: Wed Mar 2 12:47:23 2016 +0100
----------------------------------------------------------------------
.../storm/tests/StormFieldsGroupingITCase.java | 12 ++++++++---
.../apache/flink/streaming/api/IterateTest.java | 9 ++++++++
.../streaming/api/StreamingOperatorsITCase.java | 22 ++++++++++++--------
.../api/scala/StreamingOperatorsITCase.scala | 4 +++-
4 files changed, 34 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b862fd0b/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/StormFieldsGroupingITCase.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/StormFieldsGroupingITCase.java b/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/StormFieldsGroupingITCase.java
index dfadd77..b873345 100644
--- a/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/StormFieldsGroupingITCase.java
+++ b/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/StormFieldsGroupingITCase.java
@@ -21,13 +21,19 @@ import backtype.storm.Config;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
+import org.apache.flink.runtime.util.MathUtils;
import org.apache.flink.storm.api.FlinkLocalCluster;
import org.apache.flink.storm.api.FlinkTopology;
import org.apache.flink.storm.tests.operators.FiniteRandomSpout;
import org.apache.flink.storm.tests.operators.TaskIdBolt;
import org.apache.flink.storm.util.BoltFileSink;
+import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.util.StreamingProgramTestBase;
+/**
+ * This test relies on the hash function used by the {@link DataStream#keyBy}, which is
+ * assumed to be {@link MathUtils#murmurHash}.
+ */
public class StormFieldsGroupingITCase extends StreamingProgramTestBase {
private final static String topologyId = "FieldsGrouping Test";
@@ -43,9 +49,9 @@ public class StormFieldsGroupingITCase extends StreamingProgramTestBase {
@Override
protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory("4> -1930858313\n" + "4> 1431162155\n" + "4> 1654374947\n"
- + "4> -65105105\n" + "3> -1155484576\n" + "3> 1033096058\n" + "3> -1557280266\n"
- + "3> -1728529858\n" + "3> -518907128\n" + "3> -252332814", this.resultPath);
+ compareResultsByLinesInMemory("3> -1155484576\n" + "3> 1033096058\n" + "3> -1930858313\n" +
+ "3> 1431162155\n" + "4> -1557280266\n" + "4> -1728529858\n" + "4> 1654374947\n" +
+ "4> -65105105\n" + "4> -518907128\n" + "4> -252332814\n", this.resultPath);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/b862fd0b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/IterateTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/IterateTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/IterateTest.java
index 920185a..27a1e3c 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/IterateTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/IterateTest.java
@@ -462,6 +462,15 @@ public class IterateTest extends StreamingMultipleProgramsTestBase {
assertEquals(Arrays.asList("1", "1", "2", "2", "2", "2"), TestSink.collected);
}
+ /**
+ * This test relies on the hash function used by the {@link DataStream#keyBy}, which is
+ * assumed to be {@link MathUtils#murmurHash}.
+ *
+ * For the test to pass all FlatMappers must see at least two records in the iteration,
+ * which can only be achieved if the hashed values of the input keys map to a complete
+ * congruence system. Given that the test is designed for 3 parallel FlatMapper instances
+ * keys chosen from the [1,3] range are a suitable choice.
+ */
@Test
public void testGroupByFeedback() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
http://git-wip-us.apache.org/repos/asf/flink/blob/b862fd0b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/StreamingOperatorsITCase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/StreamingOperatorsITCase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/StreamingOperatorsITCase.java
index 42febea..9530d09 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/StreamingOperatorsITCase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/StreamingOperatorsITCase.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.runtime.util.MathUtils;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SplitStream;
@@ -67,12 +68,13 @@ public class StreamingOperatorsITCase extends StreamingMultipleProgramsTestBase
* of Tuple2<Integer, Integer> is created. The stream is grouped according to the first tuple
* value. Each group is folded where the second tuple value is summed up.
*
- * @throws Exception
+ * This test relies on the hash function used by the {@link DataStream#keyBy}, which is
+ * assumed to be {@link MathUtils#murmurHash}.
*/
@Test
- public void testFoldOperation() throws Exception {
+ public void testGroupedFoldOperation() throws Exception {
int numElements = 10;
- int numKeys = 2;
+ final int numKeys = 2;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<Tuple2<Integer, Integer>> sourceStream = env.addSource(new TupleSource(numElements, numKeys));
@@ -85,9 +87,13 @@ public class StreamingOperatorsITCase extends StreamingMultipleProgramsTestBase
return accumulator + value.f1;
}
}).map(new RichMapFunction<Integer, Tuple2<Integer, Integer>>() {
+ int key = -1;
@Override
public Tuple2<Integer, Integer> map(Integer value) throws Exception {
- return new Tuple2<Integer, Integer>(getRuntimeContext().getIndexOfThisSubtask(), value);
+ if (key == -1){
+ key = MathUtils.murmurHash(value) % numKeys;
+ }
+ return new Tuple2<>(key, value);
}
}).split(new OutputSelector<Tuple2<Integer, Integer>>() {
@Override
@@ -95,7 +101,6 @@ public class StreamingOperatorsITCase extends StreamingMultipleProgramsTestBase
List<String> output = new ArrayList<>();
output.add(value.f0 + "");
-
return output;
}
});
@@ -120,7 +125,7 @@ public class StreamingOperatorsITCase extends StreamingMultipleProgramsTestBase
int counter2 = 0;
for (int i = 0; i < numElements; i++) {
- if (i % 2 == 0) {
+ if (MathUtils.murmurHash(i) % numKeys == 0) {
counter1 += i;
builder1.append(counter1 + "\n");
} else {
@@ -196,7 +201,7 @@ public class StreamingOperatorsITCase extends StreamingMultipleProgramsTestBase
@Override
public void run(SourceContext<Tuple2<Integer, NonSerializable>> ctx) throws Exception {
for (int i = 0; i < numElements; i++) {
- ctx.collect(new Tuple2<Integer, NonSerializable>(i, new NonSerializable(i)));
+ ctx.collect(new Tuple2<>(i, new NonSerializable(i)));
}
}
@@ -217,14 +222,13 @@ public class StreamingOperatorsITCase extends StreamingMultipleProgramsTestBase
@Override
public void run(SourceContext<Tuple2<Integer, Integer>> ctx) throws Exception {
for (int i = 0; i < numElements; i++) {
- Tuple2<Integer, Integer> result = new Tuple2<>(i % numKeys, i);
+ Tuple2<Integer, Integer> result = new Tuple2<>(MathUtils.murmurHash(i) % numKeys, i);
ctx.collect(result);
}
}
@Override
public void cancel() {
-
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b862fd0b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/StreamingOperatorsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/StreamingOperatorsITCase.scala b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/StreamingOperatorsITCase.scala
index fe49b1f..e573fe0 100644
--- a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/StreamingOperatorsITCase.scala
+++ b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/StreamingOperatorsITCase.scala
@@ -58,9 +58,11 @@ class StreamingOperatorsITCase extends ScalaStreamingMultipleProgramsTestBase {
* The stream is grouped by the first field. For each group, the resulting stream is folded by
* summing up the second tuple field.
*
+ * This test relies on the hash function used by the {@link DataStream#keyBy}, which is
+ * assumed to be {@link MathUtils#murmurHash}.
*/
@Test
- def testFoldOperator(): Unit = {
+ def testGroupedFoldOperator(): Unit = {
val numElements = 10
val numKeys = 2