You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by va...@apache.org on 2015/06/25 23:25:39 UTC
[1/5] flink git commit: [FLINK-2264] [gelly] moved utility comparison
methods to TestBaseUtils
Repository: flink
Updated Branches:
refs/heads/master cad85103e -> 6ab06278c
[FLINK-2264] [gelly] moved utility comparison methods to TestBaseUtils
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3e808684
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3e808684
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3e808684
Branch: refs/heads/master
Commit: 3e8086848de02131f9db5d7b5725e2267e066ff2
Parents: 872c3fc
Author: vasia <va...@apache.org>
Authored: Thu Jun 25 14:39:56 2015 +0200
Committer: vasia <va...@apache.org>
Committed: Thu Jun 25 22:58:17 2015 +0200
----------------------------------------------------------------------
.../test/GatherSumApplyConfigurationITCase.java | 5 +-
.../test/VertexCentricConfigurationITCase.java | 25 ++++---
.../graph/test/operations/CompareResults.java | 75 --------------------
.../graph/test/operations/DegreesITCase.java | 15 ++--
.../test/operations/FromCollectionITCase.java | 7 +-
.../test/operations/GraphCreationITCase.java | 10 +--
.../GraphCreationWithMapperITCase.java | 9 ++-
.../test/operations/GraphMutationsITCase.java | 40 +++++------
.../test/operations/GraphOperationsITCase.java | 22 +++---
.../test/operations/JoinWithEdgesITCase.java | 30 ++++----
.../test/operations/JoinWithVerticesITCase.java | 10 +--
.../graph/test/operations/MapEdgesITCase.java | 10 +--
.../test/operations/MapVerticesITCase.java | 10 +--
.../operations/ReduceOnEdgesMethodsITCase.java | 30 ++++----
.../ReduceOnNeighborMethodsITCase.java | 33 +++++----
.../apache/flink/test/util/TestBaseUtils.java | 53 +++++++++++++-
.../test/javaApiOperators/CoGroupITCase.java | 54 --------------
17 files changed, 176 insertions(+), 262 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3e808684/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyConfigurationITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyConfigurationITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyConfigurationITCase.java
index 4ff14a2..701eda9 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyConfigurationITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyConfigurationITCase.java
@@ -31,7 +31,6 @@ import org.apache.flink.graph.gsa.GatherFunction;
import org.apache.flink.graph.gsa.GatherSumApplyIteration;
import org.apache.flink.graph.gsa.Neighbor;
import org.apache.flink.graph.gsa.SumFunction;
-import org.apache.flink.graph.test.operations.CompareResults;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.flink.types.LongValue;
import org.junit.Assert;
@@ -81,7 +80,7 @@ public class GatherSumApplyConfigurationITCase extends MultipleProgramsTestBase
"4,11\n" +
"5,11";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -116,7 +115,7 @@ public class GatherSumApplyConfigurationITCase extends MultipleProgramsTestBase
"4,14\n" +
"5,15";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/3e808684/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/VertexCentricConfigurationITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/VertexCentricConfigurationITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/VertexCentricConfigurationITCase.java
index 5732adb..0feb3fb 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/VertexCentricConfigurationITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/VertexCentricConfigurationITCase.java
@@ -36,7 +36,6 @@ import org.apache.flink.graph.spargel.MessagingFunction;
import org.apache.flink.graph.spargel.VertexCentricConfiguration;
import org.apache.flink.graph.spargel.VertexCentricIteration;
import org.apache.flink.graph.spargel.VertexUpdateFunction;
-import org.apache.flink.graph.test.operations.CompareResults;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.flink.types.LongValue;
import org.junit.Assert;
@@ -84,7 +83,7 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
"4,11\n" +
"5,11";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -119,7 +118,7 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
"4,14\n" +
"5,15";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -146,7 +145,7 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
"4,6\n" +
"5,6";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -174,7 +173,7 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
"4,[3]\n" +
"5,[3, 4]";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -207,7 +206,7 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
"4,[5]\n" +
"5,[1]";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -240,7 +239,7 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
"4,[3, 5]\n" +
"5,[1, 3, 4]";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -265,7 +264,7 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
"4,-1\n" +
"5,-1";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -297,7 +296,7 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
"4,1\n" +
"5,2";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -322,7 +321,7 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
"4,-1\n" +
"5,-1";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -354,7 +353,7 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
"4,1\n" +
"5,1";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -379,7 +378,7 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
"4,-1\n" +
"5,-1";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -411,7 +410,7 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
"4,true\n" +
"5,true";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/3e808684/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/CompareResults.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/CompareResults.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/CompareResults.java
deleted file mode 100644
index 2372f55..0000000
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/CompareResults.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.graph.test.operations;
-
-
-import static org.junit.Assert.assertEquals;
-
-import java.util.Arrays;
-import java.util.List;
-
-import org.apache.flink.api.java.tuple.Tuple;
-
-public class CompareResults {
-
- public static <T> void compareResultAsTuples(List<T> result, String expected) {
- compareResult(result, expected, true);
- }
-
- public static <T> void compareResultAsText(List<T> result, String expected) {
- compareResult(result, expected, false);
- }
-
- private static <T> void compareResult(List<T> result, String expected, boolean asTuples) {
- String[] extectedStrings = expected.split("\n");
- String[] resultStrings = new String[result.size()];
-
- for (int i = 0; i < resultStrings.length; i++) {
- T val = result.get(i);
-
- if (asTuples) {
- if (val instanceof Tuple) {
- Tuple t = (Tuple) val;
- Object first = t.getField(0);
- StringBuilder bld = new StringBuilder(first == null ? "null" : first.toString());
- for (int pos = 1; pos < t.getArity(); pos++) {
- Object next = t.getField(pos);
- bld.append(',').append(next == null ? "null" : next.toString());
- }
- resultStrings[i] = bld.toString();
- }
- else {
- throw new IllegalArgumentException(val + " is no tuple");
- }
- }
- else {
- resultStrings[i] = (val == null) ? "null" : val.toString();
- }
- }
-
- assertEquals("Wrong number of elements result", extectedStrings.length, resultStrings.length);
-
- Arrays.sort(extectedStrings);
- Arrays.sort(resultStrings);
-
- for (int i = 0; i < extectedStrings.length; i++) {
- assertEquals(extectedStrings[i], resultStrings[i]);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/3e808684/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/DegreesITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/DegreesITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/DegreesITCase.java
index 50f97da..b2744f9 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/DegreesITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/DegreesITCase.java
@@ -60,7 +60,7 @@ public class DegreesITCase extends MultipleProgramsTestBase {
"4,1\n" +
"5,1\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@@ -85,7 +85,7 @@ public class DegreesITCase extends MultipleProgramsTestBase {
"4,1\n" +
"5,0\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -107,7 +107,7 @@ public class DegreesITCase extends MultipleProgramsTestBase {
"3,2\n" +
"4,1\n" +
"5,2\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -129,7 +129,7 @@ public class DegreesITCase extends MultipleProgramsTestBase {
"4,1\n" +
"5,3\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -151,7 +151,7 @@ public class DegreesITCase extends MultipleProgramsTestBase {
"4,2\n" +
"5,3\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -173,9 +173,6 @@ public class DegreesITCase extends MultipleProgramsTestBase {
"4,1\n" +
"5,0\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
-
-
-
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/3e808684/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/FromCollectionITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/FromCollectionITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/FromCollectionITCase.java
index 87a66ba..5a64dd7 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/FromCollectionITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/FromCollectionITCase.java
@@ -18,7 +18,6 @@
package org.apache.flink.graph.test.operations;
-
import java.util.List;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
@@ -63,7 +62,7 @@ public class FromCollectionITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -85,7 +84,7 @@ public class FromCollectionITCase extends MultipleProgramsTestBase {
"4,(null)\n" +
"5,(null)\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -107,7 +106,7 @@ public class FromCollectionITCase extends MultipleProgramsTestBase {
"4,8\n" +
"5,10\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/3e808684/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationITCase.java
index abdde0d..22a5151 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationITCase.java
@@ -63,7 +63,7 @@ public class GraphCreationITCase extends MultipleProgramsTestBase {
"4,(null)\n" +
"5,(null)\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -84,7 +84,7 @@ public class GraphCreationITCase extends MultipleProgramsTestBase {
"4,4\n" +
"5,5\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -105,7 +105,7 @@ public class GraphCreationITCase extends MultipleProgramsTestBase {
"4,(8.0,3)\n" +
"5,(10.0,4)\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -127,7 +127,7 @@ public class GraphCreationITCase extends MultipleProgramsTestBase {
result.add(res);
expectedResult = "true";
- CompareResults.compareResultAsText(result, expectedResult);
+ compareResultAsText(result, expectedResult);
}
@Test
@@ -148,7 +148,7 @@ public class GraphCreationITCase extends MultipleProgramsTestBase {
expectedResult = "false\n";
- CompareResults.compareResultAsText(result, expectedResult);
+ compareResultAsText(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/3e808684/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationWithMapperITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationWithMapperITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationWithMapperITCase.java
index 523cee5..20cbca5 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationWithMapperITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationWithMapperITCase.java
@@ -61,7 +61,7 @@ public class GraphCreationWithMapperITCase extends MultipleProgramsTestBase {
"4,0.1\n" +
"5,0.1\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -82,7 +82,7 @@ public class GraphCreationWithMapperITCase extends MultipleProgramsTestBase {
"4,(8,42)\n" +
"5,(10,42)\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -104,8 +104,7 @@ public class GraphCreationWithMapperITCase extends MultipleProgramsTestBase {
"4,0.1\n" +
"5,0.1\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
-
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -126,7 +125,7 @@ public class GraphCreationWithMapperITCase extends MultipleProgramsTestBase {
"4,(F,3)\n" +
"5,(F,4)\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/3e808684/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphMutationsITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphMutationsITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphMutationsITCase.java
index c6a8012..d6e5a9c 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphMutationsITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphMutationsITCase.java
@@ -65,7 +65,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"5,5\n" +
"6,6\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -96,7 +96,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"6,6\n" +
"7,7\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -120,7 +120,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"4,4\n" +
"5,5\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -149,7 +149,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"4,4\n" +
"5,5\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -179,7 +179,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"5,5\n" +
"6,6\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -202,7 +202,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"2,3,23\n" +
"3,4,34\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -229,7 +229,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -255,7 +255,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -282,7 +282,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -312,7 +312,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -340,7 +340,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"4,4\n" +
"5,5\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -368,7 +368,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"5,1,51\n" +
"6,1,61\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -401,7 +401,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -432,7 +432,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -460,7 +460,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -485,7 +485,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -513,7 +513,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -542,7 +542,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -568,7 +568,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -598,6 +598,6 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/3e808684/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphOperationsITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphOperationsITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphOperationsITCase.java
index 904b58a..6e5596b 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphOperationsITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphOperationsITCase.java
@@ -65,7 +65,7 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
"4,5,45\n" + "5,4,45\n" +
"5,1,51\n" + "1,5,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -89,7 +89,7 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
"5,4,45\n" +
"1,5,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
@@ -119,7 +119,7 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
expectedResult = "3,5,35\n" +
"4,5,45\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
@@ -145,7 +145,7 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
@@ -171,7 +171,7 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -189,7 +189,7 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
expectedResult = "5";
- CompareResults.compareResultAsText(result, expectedResult);
+ compareResultAsText(result, expectedResult);
}
@Test
@@ -207,7 +207,7 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
expectedResult = "7";
- CompareResults.compareResultAsText(result, expectedResult);
+ compareResultAsText(result, expectedResult);
}
@Test
@@ -225,7 +225,7 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
expectedResult = "1\n2\n3\n4\n5\n";
- CompareResults.compareResultAsText(result, expectedResult);
+ compareResultAsText(result, expectedResult);
}
@Test
@@ -246,7 +246,7 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
"3,5\n" + "4,5\n" +
"5,1\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -279,7 +279,7 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
"5,1,51\n" +
"6,1,61\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -300,6 +300,6 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
"3,5,3,5,35\n" + "4,5,4,5,45\n" +
"5,1,5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/3e808684/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithEdgesITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithEdgesITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithEdgesITCase.java
index afdfbf2..e406ce2 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithEdgesITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithEdgesITCase.java
@@ -69,7 +69,7 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"4,5,90\n" +
"5,1,102\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -97,7 +97,7 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -125,7 +125,7 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -153,7 +153,7 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -180,7 +180,7 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -208,7 +208,7 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"4,5,90\n" +
"5,1,102\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -236,7 +236,7 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -264,7 +264,7 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -292,7 +292,7 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"4,5,120\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -319,7 +319,7 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -347,7 +347,7 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"4,5,80\n" +
"5,1,102\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -375,7 +375,7 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -403,7 +403,7 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -431,7 +431,7 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,140\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -458,7 +458,7 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/3e808684/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithVerticesITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithVerticesITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithVerticesITCase.java
index 99f893e..22a5535 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithVerticesITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithVerticesITCase.java
@@ -66,7 +66,7 @@ public class JoinWithVerticesITCase extends MultipleProgramsTestBase {
"4,8\n" +
"5,10\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -92,7 +92,7 @@ public class JoinWithVerticesITCase extends MultipleProgramsTestBase {
"4,4\n" +
"5,5\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -118,7 +118,7 @@ public class JoinWithVerticesITCase extends MultipleProgramsTestBase {
"4,4\n" +
"5,5\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -144,7 +144,7 @@ public class JoinWithVerticesITCase extends MultipleProgramsTestBase {
"4,40\n" +
"5,5\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -169,7 +169,7 @@ public class JoinWithVerticesITCase extends MultipleProgramsTestBase {
"4,40\n" +
"5,5\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/3e808684/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapEdgesITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapEdgesITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapEdgesITCase.java
index b3e9b11..35f7b0e 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapEdgesITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapEdgesITCase.java
@@ -64,7 +64,7 @@ public class MapEdgesITCase extends MultipleProgramsTestBase {
"4,5,46\n" +
"5,1,52\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -88,7 +88,7 @@ public class MapEdgesITCase extends MultipleProgramsTestBase {
"4,5,string(45)\n" +
"5,1,string(51)\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -112,7 +112,7 @@ public class MapEdgesITCase extends MultipleProgramsTestBase {
"4,5,(45)\n" +
"5,1,(51)\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -136,7 +136,7 @@ public class MapEdgesITCase extends MultipleProgramsTestBase {
"4,5,(T,45)\n" +
"5,1,(T,51)\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -161,7 +161,7 @@ public class MapEdgesITCase extends MultipleProgramsTestBase {
"4,5,(45.0,45)\n" +
"5,1,(51.0,51)\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/3e808684/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapVerticesITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapVerticesITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapVerticesITCase.java
index 0c63b4e..677a03c 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapVerticesITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapVerticesITCase.java
@@ -62,7 +62,7 @@ public class MapVerticesITCase extends MultipleProgramsTestBase {
"4,5\n" +
"5,6\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -84,7 +84,7 @@ public class MapVerticesITCase extends MultipleProgramsTestBase {
"4,four\n" +
"5,five\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -106,7 +106,7 @@ public class MapVerticesITCase extends MultipleProgramsTestBase {
"4,(4)\n" +
"5,(5)\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -128,7 +128,7 @@ public class MapVerticesITCase extends MultipleProgramsTestBase {
"4,(T,4)\n" +
"5,(T,5)\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -151,7 +151,7 @@ public class MapVerticesITCase extends MultipleProgramsTestBase {
"4,(4.0,4)\n" +
"5,(5.0,5)\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/3e808684/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnEdgesMethodsITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnEdgesMethodsITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnEdgesMethodsITCase.java
index c7aa45c..3bb19fa 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnEdgesMethodsITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnEdgesMethodsITCase.java
@@ -67,7 +67,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"4,5\n" +
"5,1\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -90,7 +90,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"4,3\n" +
"5,3\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -114,7 +114,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"4,5\n" +
"5,1";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -137,7 +137,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"3,5\n" +
"4,5";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -158,7 +158,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"4,5\n" +
"5,1";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -182,7 +182,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"5,3\n" +
"5,4";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -204,7 +204,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"3,2\n" +
"4,3";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -226,7 +226,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"5,3\n" +
"5,4";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -257,7 +257,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"5,3\n" +
"5,4";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -283,7 +283,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"4,3\n" +
"4,5";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -303,7 +303,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"5,3\n" +
"5,4";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -326,7 +326,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"4,45\n" +
"5,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -349,7 +349,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"4,45\n" +
"5,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -372,7 +372,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"4,34\n" +
"5,35\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -395,7 +395,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"4,45\n" +
"5,51\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/3e808684/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnNeighborMethodsITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnNeighborMethodsITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnNeighborMethodsITCase.java
index f25391e..7553b32 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnNeighborMethodsITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnNeighborMethodsITCase.java
@@ -68,7 +68,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
"4,5\n" +
"5,1\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -91,7 +91,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
"4,102\n" +
"5,285\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@@ -117,7 +117,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
"4,12\n" +
"5,13\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -137,7 +137,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
expectedResult = "4,5\n" +
"5,1\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -157,7 +157,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
expectedResult = "4,102\n" +
"5,285\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -178,7 +178,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
expectedResult = "4,12\n" +
"5,13\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -201,7 +201,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
"4,5\n" +
"5,1\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -224,7 +224,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
"4,102\n" +
"5,285\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -247,7 +247,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
"4,8\n" +
"5,8\n";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -271,7 +271,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
"5,1\n" +
"5,2";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -295,7 +295,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
"5,570\n" +
"5,285";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -319,7 +319,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
"5,8\n" +
"5,16";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -347,7 +347,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
"5,1\n" +
"5,2";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -375,7 +375,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
"5,285\n" +
"5,284";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@Test
@@ -404,7 +404,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
"5,13\n" +
"5,18";
- CompareResults.compareResultAsTuples(result, expectedResult);
+ compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
@@ -665,5 +665,4 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
out.collect(new Tuple2<Long, Long>(vertex.getId(), sum + vertex.getValue() + 5));
}
}
-}
-
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/3e808684/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java b/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
index 92d66d9..a0db4d6 100644
--- a/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
+++ b/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
@@ -18,6 +18,7 @@
package org.apache.flink.test.util;
+import static org.junit.Assert.assertEquals;
import akka.actor.ActorRef;
import akka.dispatch.Futures;
import akka.pattern.Patterns;
@@ -25,6 +26,7 @@ import akka.util.Timeout;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
+import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.StreamingMode;
@@ -32,9 +34,9 @@ import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages;
import org.apache.hadoop.fs.FileSystem;
import org.junit.Assert;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
@@ -392,6 +394,55 @@ public class TestBaseUtils {
}
// --------------------------------------------------------------------------------------------
+ // Comparison methods for tests using collect()
+ // --------------------------------------------------------------------------------------------
+
+ public static <T> void compareResultAsTuples(List<T> result, String expected) {
+ compareResult(result, expected, true);
+ }
+
+ public static <T> void compareResultAsText(List<T> result, String expected) {
+ compareResult(result, expected, false);
+ }
+
+ private static <T> void compareResult(List<T> result, String expected, boolean asTuples) {
+ String[] extectedStrings = expected.split("\n");
+ String[] resultStrings = new String[result.size()];
+
+ for (int i = 0; i < resultStrings.length; i++) {
+ T val = result.get(i);
+
+ if (asTuples) {
+ if (val instanceof Tuple) {
+ Tuple t = (Tuple) val;
+ Object first = t.getField(0);
+ StringBuilder bld = new StringBuilder(first == null ? "null" : first.toString());
+ for (int pos = 1; pos < t.getArity(); pos++) {
+ Object next = t.getField(pos);
+ bld.append(',').append(next == null ? "null" : next.toString());
+ }
+ resultStrings[i] = bld.toString();
+ }
+ else {
+ throw new IllegalArgumentException(val + " is no tuple");
+ }
+ }
+ else {
+ resultStrings[i] = (val == null) ? "null" : val.toString();
+ }
+ }
+
+ assertEquals("Wrong number of elements result", extectedStrings.length, resultStrings.length);
+
+ Arrays.sort(extectedStrings);
+ Arrays.sort(resultStrings);
+
+ for (int i = 0; i < extectedStrings.length; i++) {
+ assertEquals(extectedStrings[i], resultStrings[i]);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
// Miscellaneous helper methods
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3e808684/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/CoGroupITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/CoGroupITCase.java b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/CoGroupITCase.java
index 115ab0d..7faa6cc 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/CoGroupITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/CoGroupITCase.java
@@ -23,7 +23,6 @@ import org.apache.flink.api.common.functions.RichCoGroupFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.KeySelector;
-import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
@@ -42,12 +41,9 @@ import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collection;
import java.util.List;
-import static org.junit.Assert.assertEquals;
-
@RunWith(Parameterized.class)
public class CoGroupITCase extends MultipleProgramsTestBase {
@@ -519,56 +515,6 @@ public class CoGroupITCase extends MultipleProgramsTestBase {
compareResultAsText(result, expected);
}
-
-
- // --------------------------------------------------------------------------------------------
- // Utilities
- // --------------------------------------------------------------------------------------------
-
- private <T> void compareResultAsTuples(List<T> result, String expected) {
- compareResult(result, expected, true);
- }
-
- private <T> void compareResultAsText(List<T> result, String expected) {
- compareResult(result, expected, false);
- }
-
- private <T> void compareResult(List<T> result, String expected, boolean asTuples) {
- String[] extectedStrings = expected.split("\n");
- String[] resultStrings = new String[result.size()];
-
- for (int i = 0; i < resultStrings.length; i++) {
- T val = result.get(i);
-
- if (asTuples) {
- if (val instanceof Tuple) {
- Tuple t = (Tuple) val;
- Object first = t.getField(0);
- StringBuilder bld = new StringBuilder(first == null ? "null" : first.toString());
- for (int pos = 1; pos < t.getArity(); pos++) {
- Object next = t.getField(pos);
- bld.append(',').append(next == null ? "null" : next.toString());
- }
- resultStrings[i] = bld.toString();
- }
- else {
- throw new IllegalArgumentException(val + " is no tuple");
- }
- }
- else {
- resultStrings[i] = (val == null) ? "null" : val.toString();
- }
- }
-
- assertEquals("Wrong number of elements result", extectedStrings.length, resultStrings.length);
-
- Arrays.sort(extectedStrings);
- Arrays.sort(resultStrings);
-
- for (int i = 0; i < extectedStrings.length; i++) {
- assertEquals(extectedStrings[i], resultStrings[i]);
- }
- }
// --------------------------------------------------------------------------------------------
// UDF classes
[3/5] flink git commit: [FLINK-2264] [gelly] changed the tests to use
collect instead of temp files
Posted by va...@apache.org.
[FLINK-2264] [gelly] changed the tests to use collect instead of temp files
This closes #863
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/872c3fc4
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/872c3fc4
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/872c3fc4
Branch: refs/heads/master
Commit: 872c3fc4e191f26e3182d33f7c6e688d406cbeb7
Parents: fe5bbdb
Author: Samia <samia>
Authored: Wed Jun 24 12:41:34 2015 +0200
Committer: vasia <va...@apache.org>
Committed: Thu Jun 25 22:58:17 2015 +0200
----------------------------------------------------------------------
.../test/GatherSumApplyConfigurationITCase.java | 56 ++---
.../test/VertexCentricConfigurationITCase.java | 211 ++++++++++---------
.../graph/test/operations/CompareResults.java | 75 +++++++
.../graph/test/operations/DegreesITCase.java | 71 ++++---
.../test/operations/FromCollectionITCase.java | 44 ++--
.../test/operations/GraphCreationITCase.java | 69 +++---
.../GraphCreationWithMapperITCase.java | 50 ++---
.../test/operations/GraphMutationsITCase.java | 153 +++++++++-----
.../test/operations/GraphOperationsITCase.java | 110 ++++++----
.../test/operations/JoinWithEdgesITCase.java | 144 +++++++------
.../test/operations/JoinWithVerticesITCase.java | 61 +++---
.../graph/test/operations/MapEdgesITCase.java | 50 ++---
.../test/operations/MapVerticesITCase.java | 53 ++---
.../operations/ReduceOnEdgesMethodsITCase.java | 96 ++++-----
.../ReduceOnNeighborMethodsITCase.java | 115 +++++-----
15 files changed, 729 insertions(+), 629 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/872c3fc4/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyConfigurationITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyConfigurationITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyConfigurationITCase.java
index 5befafe..4ff14a2 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyConfigurationITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyConfigurationITCase.java
@@ -31,14 +31,11 @@ import org.apache.flink.graph.gsa.GatherFunction;
import org.apache.flink.graph.gsa.GatherSumApplyIteration;
import org.apache.flink.graph.gsa.Neighbor;
import org.apache.flink.graph.gsa.SumFunction;
+import org.apache.flink.graph.test.operations.CompareResults;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.flink.types.LongValue;
-import org.junit.After;
import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -51,22 +48,8 @@ public class GatherSumApplyConfigurationITCase extends MultipleProgramsTestBase
super(mode);
}
- private String resultPath;
private String expectedResult;
- @Rule
- public TemporaryFolder tempFolder = new TemporaryFolder();
-
- @Before
- public void before() throws Exception{
- resultPath = tempFolder.newFile().toURI().toString();
- }
-
- @After
- public void after() throws Exception{
- compareResultsByLinesInMemory(expectedResult, resultPath);
- }
-
@Test
public void testRunWithConfiguration() throws Exception {
/*
@@ -86,17 +69,19 @@ public class GatherSumApplyConfigurationITCase extends MultipleProgramsTestBase
parameters.registerAggregator("superstepAggregator", new LongSumAggregator());
parameters.setOptNumVertices(true);
- Graph<Long, Long, Long> result = graph.runGatherSumApplyIteration(new Gather(), new Sum(),
+ Graph<Long, Long, Long> res = graph.runGatherSumApplyIteration(new Gather(), new Sum(),
new Apply(), 10, parameters);
- result.getVertices().writeAsCsv(resultPath, "\n", "\t");
- env.execute();
+ DataSet<Vertex<Long, Long>> data = res.getVertices();
+ List<Vertex<Long, Long>> result= data.collect();
- expectedResult = "1 11\n" +
- "2 11\n" +
- "3 11\n" +
- "4 11\n" +
- "5 11";
+ expectedResult = "1,11\n" +
+ "2,11\n" +
+ "3,11\n" +
+ "4,11\n" +
+ "5,11";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -122,15 +107,16 @@ public class GatherSumApplyConfigurationITCase extends MultipleProgramsTestBase
Assert.assertEquals(2, iteration.getIterationConfiguration().getParallelism());
Assert.assertEquals(true, iteration.getIterationConfiguration().isSolutionSetUnmanagedMemory());
- DataSet<Vertex<Long, Long>> result = TestGraphUtils.getLongLongVertexData(env).runOperation(iteration);
-
- result.writeAsCsv(resultPath, "\n", "\t");
- env.execute();
- expectedResult = "1 11\n" +
- "2 12\n" +
- "3 13\n" +
- "4 14\n" +
- "5 15";
+ DataSet<Vertex<Long, Long>> data = TestGraphUtils.getLongLongVertexData(env).runOperation(iteration);
+ List<Vertex<Long, Long>> result= data.collect();
+
+ expectedResult = "1,11\n" +
+ "2,12\n" +
+ "3,13\n" +
+ "4,14\n" +
+ "5,15";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/872c3fc4/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/VertexCentricConfigurationITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/VertexCentricConfigurationITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/VertexCentricConfigurationITCase.java
index 567c015..5732adb 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/VertexCentricConfigurationITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/VertexCentricConfigurationITCase.java
@@ -26,6 +26,7 @@ import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple1;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.graph.Edge;
import org.apache.flink.graph.EdgeDirection;
import org.apache.flink.graph.Graph;
@@ -35,14 +36,11 @@ import org.apache.flink.graph.spargel.MessagingFunction;
import org.apache.flink.graph.spargel.VertexCentricConfiguration;
import org.apache.flink.graph.spargel.VertexCentricIteration;
import org.apache.flink.graph.spargel.VertexUpdateFunction;
+import org.apache.flink.graph.test.operations.CompareResults;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.flink.types.LongValue;
-import org.junit.After;
import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.apache.flink.graph.utils.VertexToTuple2Map;
@@ -54,22 +52,8 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
super(mode);
}
- private String resultPath;
private String expectedResult;
- @Rule
- public TemporaryFolder tempFolder = new TemporaryFolder();
-
- @Before
- public void before() throws Exception{
- resultPath = tempFolder.newFile().toURI().toString();
- }
-
- @After
- public void after() throws Exception{
- compareResultsByLinesInMemory(expectedResult, resultPath);
- }
-
@Test
public void testRunWithConfiguration() throws Exception {
/*
@@ -88,16 +72,19 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
parameters.registerAggregator("superstepAggregator", new LongSumAggregator());
parameters.setOptNumVertices(true);
- Graph<Long, Long, Long> result = graph.runVertexCentricIteration(
+ Graph<Long, Long, Long> res = graph.runVertexCentricIteration(
new UpdateFunction(), new MessageFunction(), 10, parameters);
- result.getVertices().writeAsCsv(resultPath, "\n", "\t");
- env.execute();
- expectedResult = "1 11\n" +
- "2 11\n" +
- "3 11\n" +
- "4 11\n" +
- "5 11";
+ DataSet<Vertex<Long,Long>> data = res.getVertices();
+ List<Vertex<Long,Long>> result= data.collect();
+
+ expectedResult = "1,11\n" +
+ "2,11\n" +
+ "3,11\n" +
+ "4,11\n" +
+ "5,11";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -123,15 +110,16 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
Assert.assertEquals(2, iteration.getIterationConfiguration().getParallelism());
Assert.assertEquals(true, iteration.getIterationConfiguration().isSolutionSetUnmanagedMemory());
- DataSet<Vertex<Long, Long>> result = TestGraphUtils.getLongLongVertexData(env).runOperation(iteration);
+ DataSet<Vertex<Long, Long>> data = TestGraphUtils.getLongLongVertexData(env).runOperation(iteration);
+ List<Vertex<Long,Long>> result= data.collect();
+
+ expectedResult = "1,11\n" +
+ "2,12\n" +
+ "3,13\n" +
+ "4,14\n" +
+ "5,15";
- result.writeAsCsv(resultPath, "\n", "\t");
- env.execute();
- expectedResult = "1 11\n" +
- "2 12\n" +
- "3 13\n" +
- "4 14\n" +
- "5 15";
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -145,16 +133,20 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromCollection(TestGraphUtils.getLongLongVertices(),
TestGraphUtils.getLongLongEdges(), env).mapVertices(new AssignOneMapper());
- Graph<Long, Long, Long> result = graph.runVertexCentricIteration(
+ Graph<Long, Long, Long> res = graph.runVertexCentricIteration(
new UpdateFunctionDefault(), new MessageFunctionDefault(), 5);
- result.getVertices().map(new VertexToTuple2Map<Long, Long>()).writeAsCsv(resultPath, "\n", "\t");
- env.execute();
- expectedResult = "1 6\n" +
- "2 6\n" +
- "3 6\n" +
- "4 6\n" +
- "5 6";
+
+ DataSet<Tuple2<Long, Long>> data = res.getVertices().map(new VertexToTuple2Map<Long, Long>());
+ List<Tuple2<Long, Long>> result= data.collect();
+
+ expectedResult = "1,6\n" +
+ "2,6\n" +
+ "3,6\n" +
+ "4,6\n" +
+ "5,6";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -174,14 +166,15 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
.runVertexCentricIteration(new VertexUpdateDirection(), new IdMessengerTrg(), 5)
.getVertices();
- resultedVertices.writeAsCsv(resultPath, "\n", "\t");
- env.execute();
+ List<Vertex<Long, HashSet<Long>>> result= resultedVertices.collect();
- expectedResult = "1 [5]\n" +
- "2 [1]\n" +
- "3 [1, 2]\n" +
- "4 [3]\n" +
- "5 [3, 4]";
+ expectedResult = "1,[5]\n" +
+ "2,[1]\n" +
+ "3,[1, 2]\n" +
+ "4,[3]\n" +
+ "5,[3, 4]";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -206,14 +199,15 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
.runVertexCentricIteration(new VertexUpdateDirection(), new IdMessengerSrc(), 5, parameters)
.getVertices();
- resultedVertices.writeAsCsv(resultPath, "\n", "\t");
- env.execute();
+ List<Vertex<Long, HashSet<Long>>> result= resultedVertices.collect();
- expectedResult = "1 [2, 3]\n" +
- "2 [3]\n" +
- "3 [4, 5]\n" +
- "4 [5]\n" +
- "5 [1]";
+ expectedResult = "1,[2, 3]\n" +
+ "2,[3]\n" +
+ "3,[4, 5]\n" +
+ "4,[5]\n" +
+ "5,[1]";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -238,14 +232,15 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
.runVertexCentricIteration(new VertexUpdateDirection(), new IdMessengerAll(), 5, parameters)
.getVertices();
- resultedVertices.writeAsCsv(resultPath, "\n", "\t");
- env.execute();
+ List<Vertex<Long, HashSet<Long>>> result= resultedVertices.collect();
- expectedResult = "1 [2, 3, 5]\n" +
- "2 [1, 3]\n" +
- "3 [1, 2, 4, 5]\n" +
- "4 [3, 5]\n" +
- "5 [1, 3, 4]";
+ expectedResult = "1,[2, 3, 5]\n" +
+ "2,[1, 3]\n" +
+ "3,[1, 2, 4, 5]\n" +
+ "4,[3, 5]\n" +
+ "5,[1, 3, 4]";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -262,14 +257,15 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
DataSet<Vertex<Long, Long>> verticesWithNumVertices = graph.runVertexCentricIteration(new UpdateFunctionNumVertices(),
new DummyMessageFunction(), 2).getVertices();
- verticesWithNumVertices.writeAsCsv(resultPath, "\n", "\t");
- env.execute();
+ List<Vertex<Long, Long>> result= verticesWithNumVertices.collect();
- expectedResult = "1 -1\n" +
- "2 -1\n" +
- "3 -1\n" +
- "4 -1\n" +
- "5 -1";
+ expectedResult = "1,-1\n" +
+ "2,-1\n" +
+ "3,-1\n" +
+ "4,-1\n" +
+ "5,-1";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -293,14 +289,15 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
DataSet<Vertex<Long, Long>> verticesWithDegrees = graph.runVertexCentricIteration(
new UpdateFunctionInDegrees(), new DegreesMessageFunction(), 5, parameters).getVertices();
- verticesWithDegrees.writeAsCsv(resultPath, "\n", "\t");
- env.execute();
+ List<Vertex<Long, Long>> result= verticesWithDegrees.collect();
- expectedResult = "1 1\n" +
- "2 1\n" +
- "3 2\n" +
- "4 1\n" +
- "5 2";
+ expectedResult = "1,1\n" +
+ "2,1\n" +
+ "3,2\n" +
+ "4,1\n" +
+ "5,2";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -317,14 +314,15 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
DataSet<Vertex<Long, Long>> verticesWithDegrees = graph.runVertexCentricIteration(
new UpdateFunctionInDegrees(), new DummyMessageFunction(), 2).getVertices();
- verticesWithDegrees.writeAsCsv(resultPath, "\n", "\t");
- env.execute();
+ List<Vertex<Long, Long>> result= verticesWithDegrees.collect();
- expectedResult = "1 -1\n" +
- "2 -1\n" +
- "3 -1\n" +
- "4 -1\n" +
- "5 -1";
+ expectedResult = "1,-1\n" +
+ "2,-1\n" +
+ "3,-1\n" +
+ "4,-1\n" +
+ "5,-1";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -348,14 +346,15 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
DataSet<Vertex<Long, Long>> verticesWithDegrees = graph.runVertexCentricIteration(
new UpdateFunctionOutDegrees(), new DegreesMessageFunction(), 5, parameters).getVertices();
- verticesWithDegrees.writeAsCsv(resultPath, "\n", "\t");
- env.execute();
+ List<Vertex<Long, Long>> result= verticesWithDegrees.collect();
- expectedResult = "1 2\n" +
- "2 1\n" +
- "3 2\n" +
- "4 1\n" +
- "5 1";
+ expectedResult = "1,2\n" +
+ "2,1\n" +
+ "3,2\n" +
+ "4,1\n" +
+ "5,1";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -372,14 +371,15 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
DataSet<Vertex<Long, Long>> verticesWithDegrees = graph.runVertexCentricIteration(
new UpdateFunctionInDegrees(), new DummyMessageFunction(), 2).getVertices();
- verticesWithDegrees.writeAsCsv(resultPath, "\n", "\t");
- env.execute();
+ List<Vertex<Long, Long>> result= verticesWithDegrees.collect();
- expectedResult = "1 -1\n" +
- "2 -1\n" +
- "3 -1\n" +
- "4 -1\n" +
- "5 -1";
+ expectedResult = "1,-1\n" +
+ "2,-1\n" +
+ "3,-1\n" +
+ "4,-1\n" +
+ "5,-1";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -403,14 +403,15 @@ public class VertexCentricConfigurationITCase extends MultipleProgramsTestBase {
DataSet<Vertex<Long, Boolean>> verticesWithNumNeighbors = graph.runVertexCentricIteration(
new VertexUpdateNumNeighbors(), new IdMessenger(), 1, parameters).getVertices();
- verticesWithNumNeighbors.writeAsCsv(resultPath, "\n", "\t");
- env.execute();
+ List<Vertex<Long, Boolean>> result= verticesWithNumNeighbors.collect();
- expectedResult = "1 true\n" +
- "2 true\n" +
- "3 true\n" +
- "4 true\n" +
- "5 true";
+ expectedResult = "1,true\n" +
+ "2,true\n" +
+ "3,true\n" +
+ "4,true\n" +
+ "5,true";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/872c3fc4/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/CompareResults.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/CompareResults.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/CompareResults.java
new file mode 100644
index 0000000..2372f55
--- /dev/null
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/CompareResults.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.graph.test.operations;
+
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.flink.api.java.tuple.Tuple;
+
+public class CompareResults {
+
+ public static <T> void compareResultAsTuples(List<T> result, String expected) {
+ compareResult(result, expected, true);
+ }
+
+ public static <T> void compareResultAsText(List<T> result, String expected) {
+ compareResult(result, expected, false);
+ }
+
+ private static <T> void compareResult(List<T> result, String expected, boolean asTuples) {
+ String[] extectedStrings = expected.split("\n");
+ String[] resultStrings = new String[result.size()];
+
+ for (int i = 0; i < resultStrings.length; i++) {
+ T val = result.get(i);
+
+ if (asTuples) {
+ if (val instanceof Tuple) {
+ Tuple t = (Tuple) val;
+ Object first = t.getField(0);
+ StringBuilder bld = new StringBuilder(first == null ? "null" : first.toString());
+ for (int pos = 1; pos < t.getArity(); pos++) {
+ Object next = t.getField(pos);
+ bld.append(',').append(next == null ? "null" : next.toString());
+ }
+ resultStrings[i] = bld.toString();
+ }
+ else {
+ throw new IllegalArgumentException(val + " is no tuple");
+ }
+ }
+ else {
+ resultStrings[i] = (val == null) ? "null" : val.toString();
+ }
+ }
+
+ assertEquals("Wrong number of elements result", extectedStrings.length, resultStrings.length);
+
+ Arrays.sort(extectedStrings);
+ Arrays.sort(resultStrings);
+
+ for (int i = 0; i < extectedStrings.length; i++) {
+ assertEquals(extectedStrings[i], resultStrings[i]);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/872c3fc4/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/DegreesITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/DegreesITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/DegreesITCase.java
index 0391dcb..50f97da 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/DegreesITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/DegreesITCase.java
@@ -18,16 +18,15 @@
package org.apache.flink.graph.test.operations;
+import java.util.List;
+import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.graph.Graph;
import org.apache.flink.graph.test.TestGraphUtils;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.flink.types.NullValue;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -38,21 +37,8 @@ public class DegreesITCase extends MultipleProgramsTestBase {
super(mode);
}
- private String resultPath;
private String expectedResult;
- @Rule
- public TemporaryFolder tempFolder = new TemporaryFolder();
-
- @Before
- public void before() throws Exception{
- resultPath = tempFolder.newFile().toURI().toString();
- }
-
- @After
- public void after() throws Exception{
- compareResultsByLinesInMemory(expectedResult, resultPath);
- }
@Test
public void testOutDegrees() throws Exception {
@@ -64,14 +50,18 @@ public class DegreesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- graph.outDegrees().writeAsCsv(resultPath);
- env.execute();
-
+ DataSet<Tuple2<Long,Long>> data =graph.outDegrees();
+ List<Tuple2<Long,Long>> result= data.collect();
+
+
expectedResult = "1,2\n" +
"2,1\n" +
"3,2\n" +
"4,1\n" +
"5,1\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
+
}
@Test
@@ -84,14 +74,18 @@ public class DegreesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeDataWithZeroDegree(env), env);
- graph.outDegrees().writeAsCsv(resultPath);
- env.execute();
-
+
+
+ DataSet<Tuple2<Long,Long>> data =graph.outDegrees();
+ List<Tuple2<Long,Long>> result= data.collect();
+
expectedResult = "1,3\n" +
"2,1\n" +
"3,1\n" +
"4,1\n" +
"5,0\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -104,13 +98,16 @@ public class DegreesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- graph.inDegrees().writeAsCsv(resultPath);
- env.execute();
+
+ DataSet<Tuple2<Long,Long>> data =graph.inDegrees();
+ List<Tuple2<Long,Long>> result= data.collect();
+
expectedResult = "1,1\n" +
"2,1\n" +
"3,2\n" +
"4,1\n" +
"5,2\n";
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -123,13 +120,16 @@ public class DegreesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeDataWithZeroDegree(env), env);
- graph.inDegrees().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Tuple2<Long,Long>> data =graph.inDegrees();
+ List<Tuple2<Long,Long>> result= data.collect();
+
expectedResult = "1,0\n" +
"2,1\n" +
"3,1\n" +
"4,1\n" +
"5,3\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -142,13 +142,16 @@ public class DegreesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- graph.getDegrees().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Tuple2<Long,Long>> data =graph.getDegrees();
+ List<Tuple2<Long,Long>> result= data.collect();
+
expectedResult = "1,3\n" +
"2,2\n" +
"3,4\n" +
"4,2\n" +
"5,3\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -161,12 +164,18 @@ public class DegreesITCase extends MultipleProgramsTestBase {
Graph<Long, NullValue, Long> graph =
Graph.fromDataSet(TestGraphUtils.getDisconnectedLongLongEdgeData(env), env);
- graph.outDegrees().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Tuple2<Long,Long>> data =graph.outDegrees();
+ List<Tuple2<Long,Long>> result= data.collect();
+
expectedResult = "1,2\n" +
"2,1\n" +
"3,0\n" +
"4,1\n" +
"5,0\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
+
+
+
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/872c3fc4/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/FromCollectionITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/FromCollectionITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/FromCollectionITCase.java
index 5d4b7d7..87a66ba 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/FromCollectionITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/FromCollectionITCase.java
@@ -18,17 +18,18 @@
package org.apache.flink.graph.test.operations;
+
+import java.util.List;
import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.graph.Edge;
import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.Vertex;
import org.apache.flink.graph.test.TestGraphUtils;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.flink.types.NullValue;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -39,21 +40,8 @@ public class FromCollectionITCase extends MultipleProgramsTestBase {
super(mode);
}
- private String resultPath;
private String expectedResult;
- @Rule
- public TemporaryFolder tempFolder = new TemporaryFolder();
-
- @Before
- public void before() throws Exception{
- resultPath = tempFolder.newFile().toURI().toString();
- }
-
- @After
- public void after() throws Exception{
- compareResultsByLinesInMemory(expectedResult, resultPath);
- }
@Test
public void testFromCollectionVerticesEdges() throws Exception {
@@ -64,8 +52,9 @@ public class FromCollectionITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromCollection(TestGraphUtils.getLongLongVertices(),
TestGraphUtils.getLongLongEdges(), env);
- graph.getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = graph.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
+
expectedResult = "1,2,12\n" +
"1,3,13\n" +
"2,3,23\n" +
@@ -73,6 +62,8 @@ public class FromCollectionITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -84,13 +75,17 @@ public class FromCollectionITCase extends MultipleProgramsTestBase {
Graph<Long, NullValue, Long> graph = Graph.fromCollection(TestGraphUtils.getLongLongEdges(),
env);
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
+
+ DataSet<Vertex<Long,NullValue>> data = graph.getVertices();
+ List<Vertex<Long,NullValue>> result= data.collect();
+
expectedResult = "1,(null)\n" +
"2,(null)\n" +
"3,(null)\n" +
"4,(null)\n" +
"5,(null)\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -103,13 +98,16 @@ public class FromCollectionITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromCollection(TestGraphUtils.getLongLongEdges(),
new InitVerticesMapper(), env);
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Vertex<Long,Long>> data = graph.getVertices();
+ List<Vertex<Long,Long>> result= data.collect();
+
expectedResult = "1,2\n" +
"2,4\n" +
"3,6\n" +
"4,8\n" +
"5,10\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/872c3fc4/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationITCase.java
index 3fe69fd..abdde0d 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationITCase.java
@@ -18,6 +18,9 @@
package org.apache.flink.graph.test.operations;
+import java.util.LinkedList;
+import java.util.List;
+
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
@@ -29,11 +32,7 @@ import org.apache.flink.graph.test.TestGraphUtils.DummyCustomParameterizedType;
import org.apache.flink.graph.validation.InvalidVertexIdsValidator;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.flink.types.NullValue;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -44,21 +43,8 @@ public class GraphCreationITCase extends MultipleProgramsTestBase {
super(mode);
}
- private String resultPath;
- private String expectedResult;
-
- @Rule
- public TemporaryFolder tempFolder = new TemporaryFolder();
-
- @Before
- public void before() throws Exception{
- resultPath = tempFolder.newFile().toURI().toString();
- }
- @After
- public void after() throws Exception{
- compareResultsByLinesInMemory(expectedResult, resultPath);
- }
+ private String expectedResult;
@Test
public void testCreateWithoutVertexValues() throws Exception {
@@ -68,13 +54,16 @@ public class GraphCreationITCase extends MultipleProgramsTestBase {
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
Graph<Long, NullValue, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongEdgeData(env), env);
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Vertex<Long,NullValue>> data = graph.getVertices();
+ List<Vertex<Long,NullValue>> result= data.collect();
+
expectedResult = "1,(null)\n" +
"2,(null)\n" +
"3,(null)\n" +
"4,(null)\n" +
"5,(null)\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -86,13 +75,16 @@ public class GraphCreationITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongEdgeData(env),
new AssignIdAsValueMapper(), env);
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Vertex<Long,Long>> data = graph.getVertices();
+ List<Vertex<Long,Long>> result= data.collect();
+
expectedResult = "1,1\n" +
"2,2\n" +
"3,3\n" +
"4,4\n" +
"5,5\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -104,13 +96,16 @@ public class GraphCreationITCase extends MultipleProgramsTestBase {
Graph<Long, DummyCustomParameterizedType<Double>, Long> graph = Graph.fromDataSet(
TestGraphUtils.getLongLongEdgeData(env), new AssignCustomVertexValueMapper(), env);
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Vertex<Long,DummyCustomParameterizedType<Double>>> data = graph.getVertices();
+ List<Vertex<Long,DummyCustomParameterizedType<Double>>> result= data.collect();
+
expectedResult = "1,(2.0,0)\n" +
"2,(4.0,1)\n" +
"3,(6.0,2)\n" +
"4,(8.0,3)\n" +
"5,(10.0,4)\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -123,12 +118,16 @@ public class GraphCreationITCase extends MultipleProgramsTestBase {
DataSet<Edge<Long, Long>> edges = TestGraphUtils.getLongLongEdgeData(env);
Graph<Long, Long, Long> graph = Graph.fromDataSet(vertices, edges, env);
- Boolean result = graph.validate(new InvalidVertexIdsValidator<Long, Long, Long>());
-
- env.fromElements(result).writeAsText(resultPath);
- env.execute();
-
- expectedResult = "true\n";
+ Boolean valid = graph.validate(new InvalidVertexIdsValidator<Long, Long, Long>());
+
+ //env.fromElements(result).writeAsText(resultPath);
+
+ String res= valid.toString();//env.fromElements(valid);
+ List<String> result= new LinkedList<String>();
+ result.add(res);
+ expectedResult = "true";
+
+ CompareResults.compareResultAsText(result, expectedResult);
}
@Test
@@ -141,11 +140,15 @@ public class GraphCreationITCase extends MultipleProgramsTestBase {
DataSet<Edge<Long, Long>> edges = TestGraphUtils.getLongLongEdgeData(env);
Graph<Long, Long, Long> graph = Graph.fromDataSet(vertices, edges, env);
- Boolean result = graph.validate(new InvalidVertexIdsValidator<Long, Long, Long>());
- env.fromElements(result).writeAsText(resultPath);
- env.execute();
+ Boolean valid = graph.validate(new InvalidVertexIdsValidator<Long, Long, Long>());
+
+ String res= valid.toString();//env.fromElements(valid);
+ List<String> result= new LinkedList<String>();
+ result.add(res);
expectedResult = "false\n";
+
+ CompareResults.compareResultAsText(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/872c3fc4/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationWithMapperITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationWithMapperITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationWithMapperITCase.java
index 23d05ac..523cee5 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationWithMapperITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphCreationWithMapperITCase.java
@@ -18,18 +18,18 @@
package org.apache.flink.graph.test.operations;
+import java.util.List;
+
import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.Vertex;
import org.apache.flink.graph.test.TestGraphUtils;
import org.apache.flink.graph.test.TestGraphUtils.DummyCustomType;
import org.apache.flink.test.util.MultipleProgramsTestBase;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -40,21 +40,8 @@ public class GraphCreationWithMapperITCase extends MultipleProgramsTestBase {
super(mode);
}
- private String resultPath;
private String expectedResult;
- @Rule
- public TemporaryFolder tempFolder = new TemporaryFolder();
-
- @Before
- public void before() throws Exception{
- resultPath = tempFolder.newFile().toURI().toString();
- }
-
- @After
- public void after() throws Exception{
- compareResultsByLinesInMemory(expectedResult, resultPath);
- }
@Test
public void testWithDoubleValueMapper() throws Exception {
@@ -65,13 +52,16 @@ public class GraphCreationWithMapperITCase extends MultipleProgramsTestBase {
Graph<Long, Double, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongEdgeData(env),
new AssignDoubleValueMapper(), env);
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Vertex<Long,Double>> data = graph.getVertices();
+ List<Vertex<Long,Double>> result= data.collect();
+
expectedResult = "1,0.1\n" +
"2,0.1\n" +
"3,0.1\n" +
"4,0.1\n" +
"5,0.1\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -83,13 +73,16 @@ public class GraphCreationWithMapperITCase extends MultipleProgramsTestBase {
Graph<Long, Tuple2<Long, Long>, Long> graph = Graph.fromDataSet(
TestGraphUtils.getLongLongEdgeData(env), new AssignTuple2ValueMapper(), env);
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Vertex<Long, Tuple2<Long, Long>>> data = graph.getVertices();
+ List<Vertex<Long, Tuple2<Long, Long>>> result= data.collect();
+
expectedResult = "1,(2,42)\n" +
"2,(4,42)\n" +
"3,(6,42)\n" +
"4,(8,42)\n" +
"5,(10,42)\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -102,13 +95,17 @@ public class GraphCreationWithMapperITCase extends MultipleProgramsTestBase {
Graph<String, Double, Long> graph = Graph.fromDataSet(TestGraphUtils.getStringLongEdgeData(env),
new AssignDoubleConstantMapper(), env);
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Vertex<String,Double>> data = graph.getVertices();
+ List<Vertex<String,Double>> result= data.collect();
+
expectedResult = "1,0.1\n" +
"2,0.1\n" +
"3,0.1\n" +
"4,0.1\n" +
"5,0.1\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
+
}
@Test
@@ -120,13 +117,16 @@ public class GraphCreationWithMapperITCase extends MultipleProgramsTestBase {
Graph<Long, DummyCustomType, Long> graph = Graph.fromDataSet(
TestGraphUtils.getLongLongEdgeData(env), new AssignCustomValueMapper(), env);
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Vertex<Long,DummyCustomType>> data = graph.getVertices();
+ List<Vertex<Long,DummyCustomType>> result= data.collect();
+
expectedResult = "1,(F,0)\n" +
"2,(F,1)\n" +
"3,(F,2)\n" +
"4,(F,3)\n" +
"5,(F,4)\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/872c3fc4/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphMutationsITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphMutationsITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphMutationsITCase.java
index 0d71b97..c6a8012 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphMutationsITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphMutationsITCase.java
@@ -21,17 +21,14 @@ package org.apache.flink.graph.test.operations;
import java.util.ArrayList;
import java.util.List;
+import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.graph.Edge;
import org.apache.flink.graph.Graph;
import org.apache.flink.graph.Vertex;
import org.apache.flink.graph.test.TestGraphUtils;
import org.apache.flink.test.util.MultipleProgramsTestBase;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -42,21 +39,8 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
super(mode);
}
- private String resultPath;
private String expectedResult;
- @Rule
- public TemporaryFolder tempFolder = new TemporaryFolder();
-
- @Before
- public void before() throws Exception{
- resultPath = tempFolder.newFile().toURI().toString();
- }
-
- @After
- public void after() throws Exception{
- compareResultsByLinesInMemory(expectedResult, resultPath);
- }
@Test
public void testAddVertex() throws Exception {
@@ -70,8 +54,9 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
TestGraphUtils.getLongLongEdgeData(env), env);
graph = graph.addVertex(new Vertex<Long, Long>(6L, 6L));
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
+
+ DataSet<Vertex<Long,Long>> data = graph.getVertices();
+ List<Vertex<Long,Long>> result= data.collect();
expectedResult = "1,1\n" +
"2,2\n" +
@@ -79,6 +64,8 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"4,4\n" +
"5,5\n" +
"6,6\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -98,8 +85,8 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
graph = graph.addVertices(vertices);
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Vertex<Long,Long>> data = graph.getVertices();
+ List<Vertex<Long,Long>> result= data.collect();
expectedResult = "1,1\n" +
"2,2\n" +
@@ -108,6 +95,8 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"5,5\n" +
"6,6\n" +
"7,7\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -121,14 +110,17 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
TestGraphUtils.getLongLongEdgeData(env), env);
graph = graph.addVertex(new Vertex<Long, Long>(1L, 1L));
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
+
+ DataSet<Vertex<Long,Long>> data = graph.getVertices();
+ List<Vertex<Long,Long>> result= data.collect();
expectedResult = "1,1\n" +
"2,2\n" +
"3,3\n" +
"4,4\n" +
"5,5\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -148,14 +140,16 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
graph = graph.addVertices(vertices);
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Vertex<Long,Long>> data = graph.getVertices();
+ List<Vertex<Long,Long>> result= data.collect();
expectedResult = "1,1\n" +
"2,2\n" +
"3,3\n" +
"4,4\n" +
"5,5\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -175,8 +169,8 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
graph = graph.addVertices(vertices);
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Vertex<Long,Long>> data = graph.getVertices();
+ List<Vertex<Long,Long>> result= data.collect();
expectedResult = "1,1\n" +
"2,2\n" +
@@ -184,6 +178,8 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"4,4\n" +
"5,5\n" +
"6,6\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -197,13 +193,16 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
graph = graph.removeVertex(new Vertex<Long, Long>(5L, 5L));
- graph.getEdges().writeAsCsv(resultPath);
- env.execute();
+
+ DataSet<Edge<Long,Long>> data = graph.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,12\n" +
"1,3,13\n" +
"2,3,23\n" +
"3,4,34\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -222,12 +221,15 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
verticesToBeRemoved.add(new Vertex<Long, Long>(2L, 2L));
graph = graph.removeVertices(verticesToBeRemoved);
- graph.getEdges().writeAsCsv(resultPath);
- env.execute();
+
+ DataSet<Edge<Long,Long>> data = graph.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "3,4,34\n" +
"3,5,35\n" +
"4,5,45\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -241,8 +243,9 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
graph = graph.removeVertex(new Vertex<Long, Long>(6L, 6L));
- graph.getEdges().writeAsCsv(resultPath);
- env.execute();
+
+ DataSet<Edge<Long,Long>> data = graph.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,12\n" +
"1,3,13\n" +
@@ -251,6 +254,8 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -268,13 +273,16 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
verticesToBeRemoved.add(new Vertex<Long, Long>(7L, 7L));
graph = graph.removeVertices(verticesToBeRemoved);
- graph.getEdges().writeAsCsv(resultPath);
- env.execute();
+
+ DataSet<Edge<Long,Long>> data = graph.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "2,3,23\n" +
"3,4,34\n" +
"3,5,35\n" +
"4,5,45\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -292,8 +300,9 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
verticesToBeRemoved.add(new Vertex<Long, Long>(7L, 7L));
graph = graph.removeVertices(verticesToBeRemoved);
- graph.getEdges().writeAsCsv(resultPath);
- env.execute();
+
+ DataSet<Edge<Long,Long>> data = graph.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,12\n" +
"1,3,13\n" +
@@ -302,6 +311,8 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -319,14 +330,17 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
verticesToBeRemoved.add(new Vertex<Long, Long>(7L, 7L));
graph = graph.removeVertices(verticesToBeRemoved);
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
+
+ DataSet<Vertex<Long,Long>> data = graph.getVertices();
+ List<Vertex<Long, Long>> result= data.collect();
expectedResult = "1,1\n" +
"2,2\n" +
"3,3\n" +
"4,4\n" +
"5,5\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -341,8 +355,9 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
TestGraphUtils.getLongLongEdgeData(env), env);
graph = graph.addEdge(new Vertex<Long, Long>(6L, 6L), new Vertex<Long, Long>(1L, 1L),
61L);
- graph.getEdges().writeAsCsv(resultPath);
- env.execute();
+
+ DataSet<Edge<Long,Long>> data = graph.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,12\n" +
"1,3,13\n" +
@@ -352,6 +367,8 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n" +
"6,1,61\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -371,8 +388,8 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
graph = graph.addEdges(edgesToBeAdded);
- graph.getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = graph.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,12\n" +
"1,3,13\n" +
@@ -383,6 +400,8 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"4,1,41\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -402,8 +421,8 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
graph = graph.addEdges(edgesToBeAdded);
- graph.getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = graph.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,12\n" +
"1,3,13\n" +
@@ -412,6 +431,8 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -426,8 +447,9 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
TestGraphUtils.getLongLongEdgeData(env), env);
graph = graph.addEdge(new Vertex<Long, Long>(1L, 1L), new Vertex<Long, Long>(2L, 2L),
12L);
- graph.getEdges().writeAsCsv(resultPath);
- env.execute();
+
+ DataSet<Edge<Long,Long>> data = graph.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,12\n" +
"1,2,12\n" +
@@ -437,6 +459,8 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -450,8 +474,9 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
graph = graph.removeEdge(new Edge<Long, Long>(5L, 1L, 51L));
- graph.getEdges().writeAsCsv(resultPath);
- env.execute();
+
+ DataSet<Edge<Long,Long>> data = graph.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,12\n" +
"1,3,13\n" +
@@ -459,6 +484,8 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"3,4,34\n" +
"3,5,35\n" +
"4,5,45\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -476,14 +503,17 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
edgesToBeRemoved.add(new Edge<Long, Long>(2L, 3L, 23L));
graph = graph.removeEdges(edgesToBeRemoved);
- graph.getEdges().writeAsCsv(resultPath);
- env.execute();
+
+ DataSet<Edge<Long,Long>> data = graph.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,12\n" +
"1,3,13\n" +
"3,4,34\n" +
"3,5,35\n" +
"4,5,45\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -501,8 +531,9 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
edgesToBeRemoved.add(new Edge<Long, Long>(5L, 1L, 51L));
graph = graph.removeEdges(edgesToBeRemoved);
- graph.getEdges().writeAsCsv(resultPath);
- env.execute();
+
+ DataSet<Edge<Long,Long>> data = graph.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,12\n" +
"1,3,13\n" +
@@ -510,6 +541,8 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"3,4,34\n" +
"3,5,35\n" +
"4,5,45\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -523,8 +556,9 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
graph = graph.removeEdge(new Edge<Long, Long>(6L, 1L, 61L));
- graph.getEdges().writeAsCsv(resultPath);
- env.execute();
+
+ DataSet<Edge<Long,Long>> data = graph.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,12\n" +
"1,3,13\n" +
@@ -533,6 +567,8 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -550,8 +586,9 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
edgesToBeRemoved.add(new Edge<Long, Long>(6L, 1L, 61L));
graph = graph.removeEdges(edgesToBeRemoved);
- graph.getEdges().writeAsCsv(resultPath);
- env.execute();
+
+ DataSet<Edge<Long,Long>> data = graph.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,12\n" +
"1,3,13\n" +
@@ -560,5 +597,7 @@ public class GraphMutationsITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/872c3fc4/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphOperationsITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphOperationsITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphOperationsITCase.java
index 1b9d5ac..904b58a 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphOperationsITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphOperationsITCase.java
@@ -22,17 +22,16 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.graph.Edge;
import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.Triplet;
import org.apache.flink.graph.Vertex;
import org.apache.flink.graph.test.TestGraphUtils;
import org.apache.flink.test.util.MultipleProgramsTestBase;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -43,22 +42,8 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
super(mode);
}
- private String resultPath;
private String expectedResult;
- @Rule
- public TemporaryFolder tempFolder = new TemporaryFolder();
-
- @Before
- public void before() throws Exception{
- resultPath = tempFolder.newFile().toURI().toString();
- }
-
- @After
- public void after() throws Exception{
- compareResultsByLinesInMemory(expectedResult, resultPath);
- }
-
@Test
public void testUndirected() throws Exception {
/*
@@ -69,8 +54,9 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- graph.getUndirected().getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = graph.getUndirected().getEdges();
+ List<Edge<Long, Long>> result= data.collect();
+
expectedResult = "1,2,12\n" + "2,1,12\n" +
"1,3,13\n" + "3,1,13\n" +
"2,3,23\n" + "3,2,23\n" +
@@ -78,6 +64,8 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
"3,5,35\n" + "5,3,35\n" +
"4,5,45\n" + "5,4,45\n" +
"5,1,51\n" + "1,5,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -90,8 +78,9 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- graph.reverse().getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = graph.reverse().getEdges();
+ List<Edge<Long, Long>> result= data.collect();
+
expectedResult = "2,1,12\n" +
"3,1,13\n" +
"3,2,23\n" +
@@ -99,6 +88,8 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
"5,3,35\n" +
"5,4,45\n" +
"1,5,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
@@ -111,7 +102,8 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- graph.subgraph(new FilterFunction<Vertex<Long, Long>>() {
+
+ DataSet<Edge<Long,Long>> data= graph.subgraph(new FilterFunction<Vertex<Long, Long>>() {
public boolean filter(Vertex<Long, Long> vertex) throws Exception {
return (vertex.getValue() > 2);
}
@@ -120,11 +112,14 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
public boolean filter(Edge<Long, Long> edge) throws Exception {
return (edge.getValue() > 34);
}
- }).getEdges().writeAsCsv(resultPath);
+ }).getEdges();
- env.execute();
+ List<Edge<Long, Long>> result= data.collect();
+
expectedResult = "3,5,35\n" +
"4,5,45\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
@@ -137,16 +132,20 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- graph.filterOnVertices(new FilterFunction<Vertex<Long, Long>>() {
+
+ DataSet<Edge<Long,Long>> data = graph.filterOnVertices(new FilterFunction<Vertex<Long, Long>>() {
public boolean filter(Vertex<Long, Long> vertex) throws Exception {
return (vertex.getValue() > 2);
}
- }).getEdges().writeAsCsv(resultPath);
+ }).getEdges();
- env.execute();
+ List<Edge<Long, Long>> result= data.collect();
+
expectedResult = "3,4,34\n" +
"3,5,35\n" +
"4,5,45\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
@@ -159,16 +158,20 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- graph.filterOnEdges(new FilterFunction<Edge<Long, Long>>() {
+
+ DataSet<Edge<Long,Long>> data = graph.filterOnEdges(new FilterFunction<Edge<Long, Long>>() {
public boolean filter(Edge<Long, Long> edge) throws Exception {
return (edge.getValue() > 34);
}
- }).getEdges().writeAsCsv(resultPath);
+ }).getEdges();
- env.execute();
+ List<Edge<Long, Long>> result= data.collect();
+
expectedResult = "3,5,35\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -180,10 +183,13 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- env.fromElements(graph.numberOfVertices()).writeAsText(resultPath);
+ DataSet<Long> data = env.fromElements(graph.numberOfVertices());
- env.execute();
+ List<Long> result= data.collect();
+
expectedResult = "5";
+
+ CompareResults.compareResultAsText(result, expectedResult);
}
@Test
@@ -195,10 +201,13 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- env.fromElements(graph.numberOfEdges()).writeAsText(resultPath);
+ DataSet<Long> data = env.fromElements(graph.numberOfEdges());
- env.execute();
+ List<Long> result= data.collect();
+
expectedResult = "7";
+
+ CompareResults.compareResultAsText(result, expectedResult);
}
@Test
@@ -210,10 +219,13 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- graph.getVertexIds().writeAsText(resultPath);
-
- env.execute();
+
+ DataSet<Long> data = graph.getVertexIds();
+ List<Long> result= data.collect();
+
expectedResult = "1\n2\n3\n4\n5\n";
+
+ CompareResults.compareResultAsText(result, expectedResult);
}
@Test
@@ -225,13 +237,16 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- graph.getEdgeIds().writeAsCsv(resultPath);
-
- env.execute();
+
+ DataSet<Tuple2<Long,Long>> data = graph.getEdgeIds();
+ List<Tuple2<Long, Long>> result= data.collect();
+
expectedResult = "1,2\n" + "1,3\n" +
"2,3\n" + "3,4\n" +
"3,5\n" + "4,5\n" +
"5,1\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -252,9 +267,8 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
graph = graph.union(Graph.fromCollection(vertices, edges, env));
- graph.getEdges().writeAsCsv(resultPath);
-
- env.execute();
+ DataSet<Edge<Long,Long>> data = graph.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,12\n" +
"1,3,13\n" +
@@ -264,6 +278,8 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
"4,5,45\n" +
"5,1,51\n" +
"6,1,61\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -276,12 +292,14 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- graph.getTriplets().writeAsCsv(resultPath);
+ DataSet<Triplet<Long,Long,Long>> data = graph.getTriplets();
+ List<Triplet<Long,Long,Long>> result= data.collect();
- env.execute();
expectedResult = "1,2,1,2,12\n" + "1,3,1,3,13\n" +
"2,3,2,3,23\n" + "3,4,3,4,34\n" +
"3,5,3,5,35\n" + "4,5,4,5,45\n" +
"5,1,5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/872c3fc4/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithEdgesITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithEdgesITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithEdgesITCase.java
index c02f0bb..afdfbf2 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithEdgesITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithEdgesITCase.java
@@ -18,7 +18,10 @@
package org.apache.flink.graph.test.operations;
+import java.util.List;
+
import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
@@ -28,11 +31,7 @@ import org.apache.flink.graph.test.TestGraphUtils;
import org.apache.flink.graph.test.TestGraphUtils.DummyCustomParameterizedType;
import org.apache.flink.graph.utils.EdgeToTuple3Map;
import org.apache.flink.test.util.MultipleProgramsTestBase;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -43,22 +42,8 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
super(mode);
}
- private String resultPath;
private String expectedResult;
- @Rule
- public TemporaryFolder tempFolder = new TemporaryFolder();
-
- @Before
- public void before() throws Exception{
- resultPath = tempFolder.newFile().toURI().toString();
- }
-
- @After
- public void after() throws Exception{
- compareResultsByLinesInMemory(expectedResult, resultPath);
- }
-
@Test
public void testWithEdgesInputDataset() throws Exception {
/*
@@ -70,11 +55,11 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithEdges(graph.getEdges()
+ Graph<Long, Long, Long> res = graph.joinWithEdges(graph.getEdges()
.map(new EdgeToTuple3Map<Long, Long>()), new AddValuesMapper());
- result.getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = res.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,24\n" +
"1,3,26\n" +
@@ -83,6 +68,8 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"3,5,70\n" +
"4,5,90\n" +
"5,1,102\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -96,11 +83,11 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithEdges(graph.getEdges().first(3)
+ Graph<Long, Long, Long> res = graph.joinWithEdges(graph.getEdges().first(3)
.map(new EdgeToTuple3Map<Long, Long>()), new AddValuesMapper());
- result.getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = res.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,24\n" +
"1,3,26\n" +
@@ -109,6 +96,8 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -122,11 +111,11 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithEdges(graph.getEdges().first(3)
+ Graph<Long, Long, Long> res = graph.joinWithEdges(graph.getEdges().first(3)
.map(new BooleanEdgeValueMapper()), new DoubleIfTrueMapper());
- result.getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = res.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,24\n" +
"1,3,26\n" +
@@ -135,6 +124,8 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -148,11 +139,11 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithEdges(TestGraphUtils.getLongLongLongTuple3Data(env),
+ Graph<Long, Long, Long> res = graph.joinWithEdges(TestGraphUtils.getLongLongLongTuple3Data(env),
new DoubleValueMapper());
- result.getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = res.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,24\n" +
"1,3,26\n" +
@@ -161,6 +152,8 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -173,11 +166,11 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithEdges(TestGraphUtils.getLongLongCustomTuple3Data(env),
+ Graph<Long, Long, Long> res = graph.joinWithEdges(TestGraphUtils.getLongLongCustomTuple3Data(env),
new CustomValueMapper());
- result.getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = res.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,10\n" +
"1,3,20\n" +
@@ -186,6 +179,8 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -199,11 +194,11 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithEdgesOnSource(graph.getEdges()
+ Graph<Long, Long, Long> res = graph.joinWithEdgesOnSource(graph.getEdges()
.map(new ProjectSourceAndValueMapper()), new AddValuesMapper());
- result.getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = res.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,24\n" +
"1,3,25\n" +
@@ -212,6 +207,8 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"3,5,69\n" +
"4,5,90\n" +
"5,1,102\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -225,11 +222,11 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithEdgesOnSource(graph.getEdges().first(3)
+ Graph<Long, Long, Long> res = graph.joinWithEdgesOnSource(graph.getEdges().first(3)
.map(new ProjectSourceAndValueMapper()), new AddValuesMapper());
- result.getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = res.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,24\n" +
"1,3,25\n" +
@@ -238,6 +235,8 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -251,11 +250,11 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithEdgesOnSource(graph.getEdges().first(3)
+ Graph<Long, Long, Long> res = graph.joinWithEdgesOnSource(graph.getEdges().first(3)
.map(new ProjectSourceWithTrueMapper()), new DoubleIfTrueMapper());
- result.getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = res.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,24\n" +
"1,3,26\n" +
@@ -264,6 +263,8 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -277,11 +278,11 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithEdgesOnSource(TestGraphUtils.getLongLongTuple2SourceData(env),
+ Graph<Long, Long, Long> res = graph.joinWithEdgesOnSource(TestGraphUtils.getLongLongTuple2SourceData(env),
new DoubleValueMapper());
- result.getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = res.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,20\n" +
"1,3,20\n" +
@@ -290,6 +291,8 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"3,5,80\n" +
"4,5,120\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -302,11 +305,11 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithEdgesOnSource(TestGraphUtils.getLongCustomTuple2SourceData(env),
+ Graph<Long, Long, Long> res = graph.joinWithEdgesOnSource(TestGraphUtils.getLongCustomTuple2SourceData(env),
new CustomValueMapper());
- result.getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = res.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,10\n" +
"1,3,10\n" +
@@ -315,6 +318,8 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"3,5,40\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -328,11 +333,11 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithEdgesOnTarget(graph.getEdges()
+ Graph<Long, Long, Long> res = graph.joinWithEdgesOnTarget(graph.getEdges()
.map(new ProjectTargetAndValueMapper()), new AddValuesMapper());
- result.getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = res.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,24\n" +
"1,3,26\n" +
@@ -341,6 +346,8 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"3,5,70\n" +
"4,5,80\n" +
"5,1,102\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -354,14 +361,11 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithEdgesOnTarget(graph.getEdges().first(3)
+ Graph<Long, Long, Long> res = graph.joinWithEdgesOnTarget(graph.getEdges().first(3)
.map(new ProjectTargetAndValueMapper()), new AddValuesMapper());
- result.getEdges().writeAsCsv(resultPath);
- env.execute();
-
- result.getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = res.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,24\n" +
"1,3,26\n" +
@@ -370,6 +374,8 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -383,11 +389,11 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithEdgesOnTarget(graph.getEdges().first(3)
+ Graph<Long, Long, Long> res = graph.joinWithEdgesOnTarget(graph.getEdges().first(3)
.map(new ProjectTargetWithTrueMapper()), new DoubleIfTrueMapper());
- result.getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = res.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,24\n" +
"1,3,26\n" +
@@ -396,6 +402,8 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -409,11 +417,11 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithEdgesOnTarget(TestGraphUtils.getLongLongTuple2TargetData(env),
+ Graph<Long, Long, Long> res = graph.joinWithEdgesOnTarget(TestGraphUtils.getLongLongTuple2TargetData(env),
new DoubleValueMapper());
- result.getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = res.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,20\n" +
"1,3,40\n" +
@@ -422,6 +430,8 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n" +
"5,1,140\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -434,11 +444,11 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithEdgesOnTarget(TestGraphUtils.getLongCustomTuple2TargetData(env),
+ Graph<Long, Long, Long> res = graph.joinWithEdgesOnTarget(TestGraphUtils.getLongCustomTuple2TargetData(env),
new CustomValueMapper());
- result.getEdges().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Edge<Long,Long>> data = res.getEdges();
+ List<Edge<Long, Long>> result= data.collect();
expectedResult = "1,2,10\n" +
"1,3,20\n" +
@@ -447,6 +457,8 @@ public class JoinWithEdgesITCase extends MultipleProgramsTestBase {
"3,5,35\n" +
"4,5,45\n" +
"5,1,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
[2/5] flink git commit: [FLINK-2264] [gelly] changed the tests to use
collect instead of temp files
Posted by va...@apache.org.
http://git-wip-us.apache.org/repos/asf/flink/blob/872c3fc4/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithVerticesITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithVerticesITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithVerticesITCase.java
index 120e97a..99f893e 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithVerticesITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/JoinWithVerticesITCase.java
@@ -18,7 +18,10 @@
package org.apache.flink.graph.test.operations;
+import java.util.List;
+
import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.graph.Graph;
@@ -27,11 +30,7 @@ import org.apache.flink.graph.test.TestGraphUtils;
import org.apache.flink.graph.test.TestGraphUtils.DummyCustomParameterizedType;
import org.apache.flink.graph.utils.VertexToTuple2Map;
import org.apache.flink.test.util.MultipleProgramsTestBase;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -42,22 +41,8 @@ public class JoinWithVerticesITCase extends MultipleProgramsTestBase {
super(mode);
}
- private String resultPath;
private String expectedResult;
- @Rule
- public TemporaryFolder tempFolder = new TemporaryFolder();
-
- @Before
- public void before() throws Exception{
- resultPath = tempFolder.newFile().toURI().toString();
- }
-
- @After
- public void after() throws Exception{
- compareResultsByLinesInMemory(expectedResult, resultPath);
- }
-
@Test
public void testJoinWithVertexSet() throws Exception {
/*
@@ -69,17 +54,19 @@ public class JoinWithVerticesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithVertices(graph.getVertices()
+ Graph<Long, Long, Long> res = graph.joinWithVertices(graph.getVertices()
.map(new VertexToTuple2Map<Long, Long>()), new AddValuesMapper());
- result.getVertices().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Vertex<Long,Long>> data = res.getVertices();
+ List<Vertex<Long,Long>> result= data.collect();
expectedResult = "1,2\n" +
"2,4\n" +
"3,6\n" +
"4,8\n" +
"5,10\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -93,17 +80,19 @@ public class JoinWithVerticesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithVertices(graph.getVertices().first(3)
+ Graph<Long, Long, Long> res = graph.joinWithVertices(graph.getVertices().first(3)
.map(new VertexToTuple2Map<Long, Long>()), new AddValuesMapper());
- result.getVertices().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Vertex<Long,Long>> data = res.getVertices();
+ List<Vertex<Long,Long>> result= data.collect();
expectedResult = "1,2\n" +
"2,4\n" +
"3,6\n" +
"4,4\n" +
"5,5\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -117,17 +106,19 @@ public class JoinWithVerticesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithVertices(graph.getVertices().first(3)
+ Graph<Long, Long, Long> res = graph.joinWithVertices(graph.getVertices().first(3)
.map(new ProjectIdWithTrue()), new DoubleIfTrueMapper());
- result.getVertices().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Vertex<Long,Long>> data = res.getVertices();
+ List<Vertex<Long,Long>> result= data.collect();
expectedResult = "1,2\n" +
"2,4\n" +
"3,6\n" +
"4,4\n" +
"5,5\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -141,17 +132,19 @@ public class JoinWithVerticesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithVertices(TestGraphUtils.getLongLongTuple2Data(env),
+ Graph<Long, Long, Long> res = graph.joinWithVertices(TestGraphUtils.getLongLongTuple2Data(env),
new ProjectSecondMapper());
- result.getVertices().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Vertex<Long,Long>> data = res.getVertices();
+ List<Vertex<Long,Long>> result= data.collect();
expectedResult = "1,10\n" +
"2,20\n" +
"3,30\n" +
"4,40\n" +
"5,5\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -164,17 +157,19 @@ public class JoinWithVerticesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- Graph<Long, Long, Long> result = graph.joinWithVertices(TestGraphUtils.getLongCustomTuple2Data(env),
+ Graph<Long, Long, Long> res = graph.joinWithVertices(TestGraphUtils.getLongCustomTuple2Data(env),
new CustomValueMapper());
- result.getVertices().writeAsCsv(resultPath);
- env.execute();
+ DataSet<Vertex<Long,Long>> data = res.getVertices();
+ List<Vertex<Long,Long>> result= data.collect();
expectedResult = "1,10\n" +
"2,20\n" +
"3,30\n" +
"4,40\n" +
"5,5\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/872c3fc4/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapEdgesITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapEdgesITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapEdgesITCase.java
index d1ba9a5..b3e9b11 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapEdgesITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapEdgesITCase.java
@@ -18,6 +18,8 @@
package org.apache.flink.graph.test.operations;
+import java.util.List;
+
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
@@ -28,11 +30,7 @@ import org.apache.flink.graph.test.TestGraphUtils;
import org.apache.flink.graph.test.TestGraphUtils.DummyCustomParameterizedType;
import org.apache.flink.graph.test.TestGraphUtils.DummyCustomType;
import org.apache.flink.test.util.MultipleProgramsTestBase;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -43,22 +41,8 @@ public class MapEdgesITCase extends MultipleProgramsTestBase {
super(mode);
}
- private String resultPath;
private String expectedResult;
- @Rule
- public TemporaryFolder tempFolder = new TemporaryFolder();
-
- @Before
- public void before() throws Exception{
- resultPath = tempFolder.newFile().toURI().toString();
- }
-
- @After
- public void after() throws Exception{
- compareResultsByLinesInMemory(expectedResult, resultPath);
- }
-
@Test
public void testWithSameValue() throws Exception {
/*
@@ -70,9 +54,8 @@ public class MapEdgesITCase extends MultipleProgramsTestBase {
TestGraphUtils.getLongLongEdgeData(env), env);
DataSet<Edge<Long, Long>> mappedEdges = graph.mapEdges(new AddOneMapper()).getEdges();
-
- mappedEdges.writeAsCsv(resultPath);
- env.execute();
+ List<Edge<Long, Long>> result= mappedEdges.collect();
+
expectedResult = "1,2,13\n" +
"1,3,14\n" +
"2,3,24\n" +
@@ -80,6 +63,8 @@ public class MapEdgesITCase extends MultipleProgramsTestBase {
"3,5,36\n" +
"4,5,46\n" +
"5,1,52\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -93,9 +78,8 @@ public class MapEdgesITCase extends MultipleProgramsTestBase {
TestGraphUtils.getLongLongEdgeData(env), env);
DataSet<Edge<Long, String>> mappedEdges = graph.mapEdges(new ToStringMapper()).getEdges();
+ List<Edge<Long, String>> result= mappedEdges.collect();
- mappedEdges.writeAsCsv(resultPath);
- env.execute();
expectedResult = "1,2,string(12)\n" +
"1,3,string(13)\n" +
"2,3,string(23)\n" +
@@ -103,6 +87,8 @@ public class MapEdgesITCase extends MultipleProgramsTestBase {
"3,5,string(35)\n" +
"4,5,string(45)\n" +
"5,1,string(51)\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -116,9 +102,7 @@ public class MapEdgesITCase extends MultipleProgramsTestBase {
TestGraphUtils.getLongLongEdgeData(env), env);
DataSet<Edge<Long, Tuple1<Long>>> mappedEdges = graph.mapEdges(new ToTuple1Mapper()).getEdges();
-
- mappedEdges.writeAsCsv(resultPath);
- env.execute();
+ List<Edge<Long, Tuple1<Long>>> result= mappedEdges.collect();
expectedResult = "1,2,(12)\n" +
"1,3,(13)\n" +
@@ -127,6 +111,8 @@ public class MapEdgesITCase extends MultipleProgramsTestBase {
"3,5,(35)\n" +
"4,5,(45)\n" +
"5,1,(51)\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -140,9 +126,7 @@ public class MapEdgesITCase extends MultipleProgramsTestBase {
TestGraphUtils.getLongLongEdgeData(env), env);
DataSet<Edge<Long, DummyCustomType>> mappedEdges = graph.mapEdges(new ToCustomTypeMapper()).getEdges();
-
- mappedEdges.writeAsCsv(resultPath);
- env.execute();
+ List<Edge<Long, DummyCustomType>> result= mappedEdges.collect();
expectedResult = "1,2,(T,12)\n" +
"1,3,(T,13)\n" +
@@ -151,6 +135,8 @@ public class MapEdgesITCase extends MultipleProgramsTestBase {
"3,5,(T,35)\n" +
"4,5,(T,45)\n" +
"5,1,(T,51)\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -165,9 +151,7 @@ public class MapEdgesITCase extends MultipleProgramsTestBase {
DataSet<Edge<Long, DummyCustomParameterizedType<Double>>> mappedEdges = graph.mapEdges(
new ToCustomParametrizedTypeMapper()).getEdges();
-
- mappedEdges.writeAsCsv(resultPath);
- env.execute();
+ List<Edge<Long, DummyCustomParameterizedType<Double>>> result= mappedEdges.collect();
expectedResult = "1,2,(12.0,12)\n" +
"1,3,(13.0,13)\n" +
@@ -176,6 +160,8 @@ public class MapEdgesITCase extends MultipleProgramsTestBase {
"3,5,(35.0,35)\n" +
"4,5,(45.0,45)\n" +
"5,1,(51.0,51)\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/872c3fc4/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapVerticesITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapVerticesITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapVerticesITCase.java
index f3a63be..0c63b4e 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapVerticesITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/MapVerticesITCase.java
@@ -18,6 +18,8 @@
package org.apache.flink.graph.test.operations;
+import java.util.List;
+
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
@@ -28,11 +30,7 @@ import org.apache.flink.graph.test.TestGraphUtils;
import org.apache.flink.graph.test.TestGraphUtils.DummyCustomParameterizedType;
import org.apache.flink.graph.test.TestGraphUtils.DummyCustomType;
import org.apache.flink.test.util.MultipleProgramsTestBase;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -43,22 +41,8 @@ public class MapVerticesITCase extends MultipleProgramsTestBase {
super(mode);
}
- private String resultPath;
private String expectedResult;
- @Rule
- public TemporaryFolder tempFolder = new TemporaryFolder();
-
- @Before
- public void before() throws Exception{
- resultPath = tempFolder.newFile().toURI().toString();
- }
-
- @After
- public void after() throws Exception{
- compareResultsByLinesInMemory(expectedResult, resultPath);
- }
-
@Test
public void testWithSameValue() throws Exception {
/*
@@ -69,15 +53,16 @@ public class MapVerticesITCase extends MultipleProgramsTestBase {
Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
TestGraphUtils.getLongLongEdgeData(env), env);
- DataSet<Vertex<Long, Long>> mappedVertices = graph.mapVertices(new AddOneMapper()).getVertices();
-
- mappedVertices.writeAsCsv(resultPath);
- env.execute();
+ DataSet<Vertex<Long, Long>> mappedVertices = graph.mapVertices(new AddOneMapper()).getVertices();
+ List<Vertex<Long, Long>> result= mappedVertices.collect();
+
expectedResult = "1,2\n" +
"2,3\n" +
"3,4\n" +
"4,5\n" +
"5,6\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -91,15 +76,15 @@ public class MapVerticesITCase extends MultipleProgramsTestBase {
TestGraphUtils.getLongLongEdgeData(env), env);
DataSet<Vertex<Long, String>> mappedVertices = graph.mapVertices(new ToStringMapper()).getVertices();
-
- mappedVertices.writeAsCsv(resultPath);
- env.execute();
+ List<Vertex<Long, String>> result= mappedVertices.collect();
expectedResult = "1,one\n" +
"2,two\n" +
"3,three\n" +
"4,four\n" +
"5,five\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -113,15 +98,15 @@ public class MapVerticesITCase extends MultipleProgramsTestBase {
TestGraphUtils.getLongLongEdgeData(env), env);
DataSet<Vertex<Long, Tuple1<Long>>> mappedVertices = graph.mapVertices(new ToTuple1Mapper()).getVertices();
-
- mappedVertices.writeAsCsv(resultPath);
- env.execute();
+ List<Vertex<Long, Tuple1<Long>>> result= mappedVertices.collect();
expectedResult = "1,(1)\n" +
"2,(2)\n" +
"3,(3)\n" +
"4,(4)\n" +
"5,(5)\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -135,15 +120,15 @@ public class MapVerticesITCase extends MultipleProgramsTestBase {
TestGraphUtils.getLongLongEdgeData(env), env);
DataSet<Vertex<Long, DummyCustomType>> mappedVertices = graph.mapVertices(new ToCustomTypeMapper()).getVertices();
-
- mappedVertices.writeAsCsv(resultPath);
- env.execute();
+ List<Vertex<Long, DummyCustomType>> result= mappedVertices.collect();
expectedResult = "1,(T,1)\n" +
"2,(T,2)\n" +
"3,(T,3)\n" +
"4,(T,4)\n" +
"5,(T,5)\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -158,15 +143,15 @@ public class MapVerticesITCase extends MultipleProgramsTestBase {
DataSet<Vertex<Long, DummyCustomParameterizedType<Double>>> mappedVertices = graph.mapVertices(
new ToCustomParametrizedTypeMapper()).getVertices();
-
- mappedVertices.writeAsCsv(resultPath);
- env.execute();
+ List<Vertex<Long, DummyCustomParameterizedType<Double>>> result= mappedVertices.collect();
expectedResult = "1,(1.0,1)\n" +
"2,(2.0,2)\n" +
"3,(3.0,3)\n" +
"4,(4.0,4)\n" +
"5,(5.0,5)\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/872c3fc4/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnEdgesMethodsITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnEdgesMethodsITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnEdgesMethodsITCase.java
index b03268c..c7aa45c 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnEdgesMethodsITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnEdgesMethodsITCase.java
@@ -18,6 +18,8 @@
package org.apache.flink.graph.test.operations;
+import java.util.List;
+
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
@@ -31,11 +33,7 @@ import org.apache.flink.graph.Vertex;
import org.apache.flink.graph.test.TestGraphUtils;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.flink.util.Collector;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -46,22 +44,8 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
super(mode);
}
- private String resultPath;
private String expectedResult;
- @Rule
- public TemporaryFolder tempFolder = new TemporaryFolder();
-
- @Before
- public void before() throws Exception{
- resultPath = tempFolder.newFile().toURI().toString();
- }
-
- @After
- public void after() throws Exception{
- compareResultsByLinesInMemory(expectedResult, resultPath);
- }
-
@Test
public void testLowestWeightOutNeighbor() throws Exception {
/*
@@ -74,14 +58,16 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithLowestOutNeighbor =
graph.groupReduceOnEdges(new SelectMinWeightNeighbor(), EdgeDirection.OUT);
- verticesWithLowestOutNeighbor.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithLowestOutNeighbor.collect();
+
expectedResult = "1,2\n" +
"2,3\n" +
"3,4\n" +
"4,5\n" +
"5,1\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -96,14 +82,15 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithLowestOutNeighbor =
graph.groupReduceOnEdges(new SelectMinWeightInNeighbor(), EdgeDirection.IN);
- verticesWithLowestOutNeighbor.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithLowestOutNeighbor.collect();
expectedResult = "1,5\n" +
"2,1\n" +
"3,1\n" +
"4,3\n" +
"5,3\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -117,8 +104,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithAllOutNeighbors =
graph.groupReduceOnEdges(new SelectOutNeighbors(), EdgeDirection.OUT);
- verticesWithAllOutNeighbors.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithAllOutNeighbors.collect();
expectedResult = "1,2\n" +
"1,3\n" +
@@ -127,6 +113,8 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"3,5\n" +
"4,5\n" +
"5,1";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -140,8 +128,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithAllOutNeighbors =
graph.groupReduceOnEdges(new SelectOutNeighborsExcludeFive(), EdgeDirection.OUT);
- verticesWithAllOutNeighbors.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithAllOutNeighbors.collect();
expectedResult = "1,2\n" +
"1,3\n" +
@@ -149,6 +136,8 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"3,4\n" +
"3,5\n" +
"4,5";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -162,13 +151,14 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithAllOutNeighbors =
graph.groupReduceOnEdges(new SelectOutNeighborsValueGreaterThanTwo(), EdgeDirection.OUT);
- verticesWithAllOutNeighbors.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithAllOutNeighbors.collect();
expectedResult = "3,4\n" +
"3,5\n" +
"4,5\n" +
"5,1";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -182,8 +172,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithAllInNeighbors =
graph.groupReduceOnEdges(new SelectInNeighbors(), EdgeDirection.IN);
- verticesWithAllInNeighbors.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithAllInNeighbors.collect();
expectedResult = "1,5\n" +
"2,1\n" +
@@ -192,6 +181,8 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"4,3\n" +
"5,3\n" +
"5,4";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -205,14 +196,15 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithAllInNeighbors =
graph.groupReduceOnEdges(new SelectInNeighborsExceptFive(), EdgeDirection.IN);
- verticesWithAllInNeighbors.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithAllInNeighbors.collect();
expectedResult = "1,5\n" +
"2,1\n" +
"3,1\n" +
"3,2\n" +
"4,3";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -226,14 +218,15 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithAllInNeighbors =
graph.groupReduceOnEdges(new SelectInNeighborsValueGreaterThanTwo(), EdgeDirection.IN);
- verticesWithAllInNeighbors.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithAllInNeighbors.collect();
expectedResult = "3,1\n" +
"3,2\n" +
"4,3\n" +
"5,3\n" +
"5,4";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -247,8 +240,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithAllNeighbors =
graph.groupReduceOnEdges(new SelectNeighbors(), EdgeDirection.ALL);
- verticesWithAllNeighbors.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithAllNeighbors.collect();
expectedResult = "1,2\n" +
"1,3\n" +
@@ -264,6 +256,8 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"5,1\n" +
"5,3\n" +
"5,4";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -277,8 +271,7 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithAllNeighbors =
graph.groupReduceOnEdges(new SelectNeighborsExceptFiveAndTwo(), EdgeDirection.ALL);
- verticesWithAllNeighbors.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithAllNeighbors.collect();
expectedResult = "1,2\n" +
"1,3\n" +
@@ -289,6 +282,8 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
"3,5\n" +
"4,3\n" +
"4,5";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -302,12 +297,13 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithAllNeighbors =
graph.groupReduceOnEdges(new SelectNeighborsValueGreaterThanFour(), EdgeDirection.ALL);
- verticesWithAllNeighbors.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithAllNeighbors.collect();
expectedResult = "5,1\n" +
"5,3\n" +
"5,4";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -322,14 +318,15 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithMaxEdgeWeight =
graph.groupReduceOnEdges(new SelectMaxWeightNeighbor(), EdgeDirection.ALL);
- verticesWithMaxEdgeWeight.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithMaxEdgeWeight.collect();
expectedResult = "1,51\n" +
"2,23\n" +
"3,35\n" +
"4,45\n" +
"5,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -344,14 +341,15 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithLowestOutNeighbor =
graph.reduceOnEdges(new SelectMinWeightNeighborNoValue(), EdgeDirection.OUT);
- verticesWithLowestOutNeighbor.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithLowestOutNeighbor.collect();
expectedResult = "1,12\n" +
"2,23\n" +
"3,34\n" +
"4,45\n" +
"5,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -366,14 +364,15 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithLowestOutNeighbor =
graph.reduceOnEdges(new SelectMinWeightNeighborNoValue(), EdgeDirection.IN);
- verticesWithLowestOutNeighbor.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithLowestOutNeighbor.collect();
expectedResult = "1,51\n" +
"2,12\n" +
"3,13\n" +
"4,34\n" +
"5,35\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -388,14 +387,15 @@ public class ReduceOnEdgesMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithMaxEdgeWeight =
graph.reduceOnEdges(new SelectMaxWeightNeighborNoValue(), EdgeDirection.ALL);
- verticesWithMaxEdgeWeight.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithMaxEdgeWeight.collect();
expectedResult = "1,51\n" +
"2,23\n" +
"3,35\n" +
"4,45\n" +
"5,51\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
http://git-wip-us.apache.org/repos/asf/flink/blob/872c3fc4/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnNeighborMethodsITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnNeighborMethodsITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnNeighborMethodsITCase.java
index a9fb06e..f25391e 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnNeighborMethodsITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/ReduceOnNeighborMethodsITCase.java
@@ -19,6 +19,7 @@
package org.apache.flink.graph.test.operations;
import java.util.Iterator;
+import java.util.List;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
@@ -34,11 +35,7 @@ import org.apache.flink.graph.Vertex;
import org.apache.flink.graph.test.TestGraphUtils;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.flink.util.Collector;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -49,22 +46,8 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
super(mode);
}
- private String resultPath;
private String expectedResult;
- @Rule
- public TemporaryFolder tempFolder = new TemporaryFolder();
-
- @Before
- public void before() throws Exception{
- resultPath = tempFolder.newFile().toURI().toString();
- }
-
- @After
- public void after() throws Exception{
- compareResultsByLinesInMemory(expectedResult, resultPath);
- }
-
@Test
public void testSumOfOutNeighbors() throws Exception {
/*
@@ -77,14 +60,15 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithSumOfOutNeighborValues =
graph.groupReduceOnNeighbors(new SumOutNeighbors(), EdgeDirection.OUT);
-
- verticesWithSumOfOutNeighborValues.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithSumOfOutNeighborValues.collect();
+
expectedResult = "1,5\n" +
"2,3\n" +
"3,9\n" +
"4,5\n" +
"5,1\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -99,14 +83,17 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithSum =
graph.groupReduceOnNeighbors(new SumInNeighbors(), EdgeDirection.IN);
-
- verticesWithSum.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithSum.collect();
+
expectedResult = "1,255\n" +
"2,12\n" +
"3,59\n" +
"4,102\n" +
"5,285\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
+
+
}
@Test
@@ -122,15 +109,15 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithSumOfOutNeighborValues =
graph.groupReduceOnNeighbors(new SumAllNeighbors(), EdgeDirection.ALL);
-
- verticesWithSumOfOutNeighborValues.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithSumOfOutNeighborValues.collect();
expectedResult = "1,11\n" +
"2,6\n" +
"3,15\n" +
"4,12\n" +
"5,13\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -145,11 +132,12 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithSumOfOutNeighborValues =
graph.groupReduceOnNeighbors(new SumOutNeighborsIdGreaterThanThree(), EdgeDirection.OUT);
-
- verticesWithSumOfOutNeighborValues.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithSumOfOutNeighborValues.collect();
+
expectedResult = "4,5\n" +
"5,1\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -164,11 +152,12 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithSum =
graph.groupReduceOnNeighbors(new SumInNeighborsIdGreaterThanThree(), EdgeDirection.IN);
-
- verticesWithSum.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithSum.collect();
+
expectedResult = "4,102\n" +
"5,285\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -184,12 +173,12 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithSumOfOutNeighborValues =
graph.groupReduceOnNeighbors(new SumAllNeighborsIdGreaterThanThree(), EdgeDirection.ALL);
-
- verticesWithSumOfOutNeighborValues.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithSumOfOutNeighborValues.collect();
expectedResult = "4,12\n" +
"5,13\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -204,15 +193,15 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithSumOfOutNeighborValues =
graph.reduceOnNeighbors(new SumNeighbors(), EdgeDirection.OUT);
-
- verticesWithSumOfOutNeighborValues.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithSumOfOutNeighborValues.collect();
expectedResult = "1,5\n" +
"2,3\n" +
"3,9\n" +
"4,5\n" +
"5,1\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -227,14 +216,15 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithSum =
graph.groupReduceOnNeighbors(new SumInNeighborsNoValue(), EdgeDirection.IN);
- verticesWithSum.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithSum.collect();
expectedResult = "1,255\n" +
"2,12\n" +
"3,59\n" +
"4,102\n" +
"5,285\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -249,15 +239,15 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithSumOfAllNeighborValues =
graph.reduceOnNeighbors(new SumNeighbors(), EdgeDirection.ALL);
-
- verticesWithSumOfAllNeighborValues.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithSumOfAllNeighborValues.collect();
expectedResult = "1,10\n" +
"2,4\n" +
"3,12\n" +
"4,8\n" +
"5,8\n";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -272,9 +262,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithSumOfOutNeighborValues =
graph.groupReduceOnNeighbors(new SumOutNeighborsNoValueMultipliedByTwoIdGreaterThanTwo(), EdgeDirection.OUT);
-
- verticesWithSumOfOutNeighborValues.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithSumOfOutNeighborValues.collect();
expectedResult = "3,9\n" +
"3,18\n" +
@@ -282,6 +270,8 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
"4,10\n" +
"5,1\n" +
"5,2";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -296,8 +286,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithSumOfOutNeighborValues =
graph.groupReduceOnNeighbors(new SumInNeighborsNoValueMultipliedByTwoIdGreaterThanTwo(), EdgeDirection.IN);
- verticesWithSumOfOutNeighborValues.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithSumOfOutNeighborValues.collect();
expectedResult = "3,59\n" +
"3,118\n" +
@@ -305,6 +294,8 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
"4,102\n" +
"5,570\n" +
"5,285";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -319,9 +310,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithSumOfAllNeighborValues =
graph.groupReduceOnNeighbors(new SumAllNeighborsNoValueMultipliedByTwoIdGreaterThanTwo(), EdgeDirection.ALL);
-
- verticesWithSumOfAllNeighborValues.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithSumOfAllNeighborValues.collect();
expectedResult = "3,12\n" +
"3,24\n" +
@@ -329,6 +318,8 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
"4,16\n" +
"5,8\n" +
"5,16";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -343,9 +334,8 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithSumOfOutNeighborValues =
graph.groupReduceOnNeighbors(new SumOutNeighborsMultipliedByTwo(), EdgeDirection.OUT);
-
- verticesWithSumOfOutNeighborValues.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithSumOfOutNeighborValues.collect();
+
expectedResult = "1,5\n" +
"1,10\n" +
"2,3\n" +
@@ -356,6 +346,8 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
"4,10\n" +
"5,1\n" +
"5,2";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -370,9 +362,8 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithSum =
graph.groupReduceOnNeighbors(new SumInNeighborsSubtractOne(), EdgeDirection.IN);
-
- verticesWithSum.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithSum.collect();
+
expectedResult = "1,255\n" +
"1,254\n" +
"2,12\n" +
@@ -383,6 +374,8 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
"4,101\n" +
"5,285\n" +
"5,284";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@Test
@@ -398,9 +391,7 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
DataSet<Tuple2<Long, Long>> verticesWithSumOfOutNeighborValues =
graph.groupReduceOnNeighbors(new SumAllNeighborsAddFive(), EdgeDirection.ALL);
-
- verticesWithSumOfOutNeighborValues.writeAsCsv(resultPath);
- env.execute();
+ List<Tuple2<Long,Long>> result = verticesWithSumOfOutNeighborValues.collect();
expectedResult = "1,11\n" +
"1,16\n" +
@@ -412,6 +403,8 @@ public class ReduceOnNeighborMethodsITCase extends MultipleProgramsTestBase {
"4,17\n" +
"5,13\n" +
"5,18";
+
+ CompareResults.compareResultAsTuples(result, expectedResult);
}
@SuppressWarnings("serial")
[5/5] flink git commit: [FLINK-2093] [gelly] Added difference Method
Posted by va...@apache.org.
[FLINK-2093] [gelly] Added difference Method
changed difference method tests to use collect()
This closes #818
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6ab06278
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6ab06278
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6ab06278
Branch: refs/heads/master
Commit: 6ab06278cdc394c009a715ac64ed06f509c7acdc
Parents: 3e80868
Author: Shivani <sh...@gmail.com>
Authored: Wed Jun 10 13:22:37 2015 +0200
Committer: vasia <va...@apache.org>
Committed: Thu Jun 25 22:58:18 2015 +0200
----------------------------------------------------------------------
docs/libs/gelly_guide.md | 8 ++-
.../main/java/org/apache/flink/graph/Graph.java | 31 +++++++-
.../apache/flink/graph/test/TestGraphUtils.java | 44 ++++++++++++
.../test/operations/GraphOperationsITCase.java | 76 +++++++++++++++++++-
4 files changed, 153 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6ab06278/docs/libs/gelly_guide.md
----------------------------------------------------------------------
diff --git a/docs/libs/gelly_guide.md b/docs/libs/gelly_guide.md
index eb10405..1e9bf48 100644
--- a/docs/libs/gelly_guide.md
+++ b/docs/libs/gelly_guide.md
@@ -236,13 +236,15 @@ Graph<Long, Double, Double> networkWithWeights = network.joinWithEdgesOnSource(v
* <strong>Undirected</strong>: In Gelly, a `Graph` is always directed. Undirected graphs can be represented by adding all opposite-direction edges to a graph. For this purpose, Gelly provides the `getUndirected()` method.
-* <strong>Union</strong>: Gelly's `union()` method performs a union on the vertex and edges sets of the input graphs. Duplicate vertices are removed from the resulting `Graph`, while if duplicate edges exists, these will be maintained.
+* <strong>Union</strong>: Gelly's `union()` method performs a union operation on the vertex and edge sets of the specified graph and current graph. Duplicate vertices are removed from the resulting `Graph`, while if duplicate edges exists, these will be maintained.
<p class="text-center">
<img alt="Union Transformation" width="50%" src="fig/gelly-union.png"/>
</p>
-[Back to top](#top)
+* <strong>Difference</strong>: Gelly's `difference()` method performs a difference on the vertex and edge sets of the current graph and specified graph.
+
+-[Back to top](#top)
Graph Mutations
-----------
@@ -273,6 +275,8 @@ Graph<K, VV, EV> removeEdge(Edge<K, EV> edge)
// removes *all* edges that match the edges in the given list
Graph<K, VV, EV> removeEdges(List<Edge<K, EV>> edgesToBeRemoved)
+
+
{% endhighlight %}
Neighborhood Methods
http://git-wip-us.apache.org/repos/asf/flink/blob/6ab06278/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/Graph.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/Graph.java b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/Graph.java
index a7c75bc..ff27949 100755
--- a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/Graph.java
+++ b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/Graph.java
@@ -1119,7 +1119,6 @@ public class Graph<K, VV, EV> {
return removeVertices(vertexToBeRemoved);
}
-
/**
* Removes the given list of vertices and its edges from the graph.
*
@@ -1127,9 +1126,22 @@ public class Graph<K, VV, EV> {
* @return the resulted graph containing the initial vertices and edges minus the vertices
* and edges removed.
*/
- public Graph<K, VV, EV> removeVertices(List<Vertex<K, VV>> verticesToBeRemoved) {
- DataSet<Vertex<K, VV>> newVertices = getVertices().coGroup(this.context.fromCollection(verticesToBeRemoved)).where(0).equalTo(0)
+ public Graph<K, VV, EV> removeVertices(List<Vertex<K, VV>> verticesToBeRemoved)
+ {
+ return removeVertices(this.context.fromCollection(verticesToBeRemoved));
+ }
+
+ /**
+ * Removes the given list of vertices and its edges from the graph.
+ *
+ * @param verticesToBeRemoved the DataSet of vertices to be removed
+ * @return the resulted graph containing the initial vertices and edges minus the vertices
+ * and edges removed.
+ */
+ private Graph<K, VV, EV> removeVertices(DataSet<Vertex<K, VV>> verticesToBeRemoved) {
+
+ DataSet<Vertex<K, VV>> newVertices = getVertices().coGroup(verticesToBeRemoved).where(0).equalTo(0)
.with(new VerticesRemovalCoGroup<K, VV>());
DataSet < Edge < K, EV >> newEdges = newVertices.join(getEdges()).where(0).equalTo(0)
@@ -1161,6 +1173,8 @@ public class Graph<K, VV, EV> {
}
}
+
+
@ForwardedFieldsSecond("f0; f1; f2")
private static final class ProjectEdgeToBeRemoved<K,VV,EV> implements JoinFunction<Vertex<K, VV>, Edge<K, EV>, Edge<K, EV>> {
@Override
@@ -1244,6 +1258,17 @@ public class Graph<K, VV, EV> {
}
/**
+ * Performs Difference on the vertex and edge sets of the input graphs
+ * removes common vertices and edges. If a source/target vertex is removed, its corresponding edge will also be removed
+ * @param graph the graph to perform difference with
+ * @return a new graph where the common vertices and edges have been removed
+ */
+ public Graph<K,VV,EV> difference(Graph<K,VV,EV> graph) {
+ DataSet<Vertex<K,VV>> removeVerticesData = graph.getVertices();
+ return this.removeVertices(removeVerticesData);
+ }
+
+ /**
* Runs a Vertex-Centric iteration on the graph.
* No configuration options are provided.
*
http://git-wip-us.apache.org/repos/asf/flink/blob/6ab06278/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphUtils.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphUtils.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphUtils.java
index 1769a26..51b5b7f 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphUtils.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphUtils.java
@@ -381,4 +381,48 @@ public class TestGraphUtils {
@Override
public void write(int b){}
}
+
+ /**
+ * utils for getting the second graph for the test of method difference();
+ * @param env
+ */
+ public static final DataSet<Edge<Long,Long>> getLongLongEdgeDataDifference(
+ ExecutionEnvironment env){
+ return env.fromCollection(getLongLongEdgesForDifference());
+ }
+
+ public static final DataSet<Edge<Long,Long>> getLongLongEdgeDataDifference2(
+ ExecutionEnvironment env){
+ return env.fromCollection(getLongLongEdgesForDifference2());
+ }
+
+ public static final DataSet<Vertex<Long,Long>> getLongLongVertexDataDifference(
+ ExecutionEnvironment env)
+ {
+ return env.fromCollection(getVerticesForDifference());
+ }
+
+ public static final List<Vertex<Long,Long>> getVerticesForDifference(){
+ List<Vertex<Long,Long>> vertices = new ArrayList<Vertex<Long,Long>>();
+ vertices.add(new Vertex<Long, Long>(1L, 1L));
+ vertices.add(new Vertex<Long, Long>(3L, 3L));
+ vertices.add(new Vertex<Long, Long>(6L, 6L));
+
+ return vertices;
+
+ }
+
+ public static final List<Edge<Long, Long>> getLongLongEdgesForDifference() {
+ List<Edge<Long, Long>> edges = new ArrayList<Edge<Long, Long>>();
+ edges.add(new Edge<Long, Long>(1L, 3L, 13L));
+ edges.add(new Edge<Long, Long>(1L, 6L, 26L));
+ edges.add(new Edge<Long, Long>(6L, 3L, 63L));
+ return edges;
+ }
+
+ public static final List<Edge<Long, Long>> getLongLongEdgesForDifference2() {
+ List<Edge<Long, Long>> edges = new ArrayList<Edge<Long, Long>>();
+ edges.add(new Edge<Long, Long>(6L, 6L, 66L));
+ return edges;
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/6ab06278/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphOperationsITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphOperationsITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphOperationsITCase.java
index 6e5596b..a06f881 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphOperationsITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/operations/GraphOperationsITCase.java
@@ -165,7 +165,7 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
}
}).getEdges();
- List<Edge<Long, Long>> result= data.collect();
+ List<Edge<Long, Long>> result = data.collect();
expectedResult = "3,5,35\n" +
"4,5,45\n" +
@@ -283,6 +283,80 @@ public class GraphOperationsITCase extends MultipleProgramsTestBase {
}
@Test
+ public void testDifference() throws Exception {
+ /*Test difference() method by checking the output for getEdges() on the resultant graph
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+
+ Graph<Long, Long, Long> graph2 = Graph.fromDataSet(TestGraphUtils.getLongLongVertexDataDifference(env),
+ TestGraphUtils.getLongLongEdgeDataDifference(env), env);
+
+ graph = graph.difference(graph2);
+
+ List<Edge<Long, Long>> result = graph.getEdges().collect();
+
+ expectedResult = "4,5,45\n";
+ compareResultAsTuples(result, expectedResult);
+ }
+
+
+ @Test
+ public void testDifferenceVertices() throws Exception{
+ /*Test difference() method by checking the output for getVertices() on the resultant graph
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+
+ Graph<Long, Long, Long> graph2 = Graph.fromDataSet(TestGraphUtils.getLongLongVertexDataDifference(env),
+ TestGraphUtils.getLongLongEdgeDataDifference(env), env);
+
+ graph = graph.difference(graph2);
+
+ List<Vertex<Long, Long>> result = graph.getVertices().collect();
+
+ expectedResult = "2,2\n" +
+ "4,4\n" +
+ "5,5\n" ;
+
+ compareResultAsTuples(result, expectedResult);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testDifference2() throws Exception {
+ /*
+ * Test difference() such that no common vertices are there
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+
+ DataSet<Vertex<Long, Long>> vertex = env.fromElements(new Vertex<Long, Long>(6L, 6L));
+
+ Graph<Long, Long, Long> graph2 = Graph.fromDataSet(vertex,TestGraphUtils.getLongLongEdgeDataDifference2(env),env);
+
+ graph = graph.difference(graph2);
+
+ List<Edge<Long, Long>> result = graph.getEdges().collect();
+
+ expectedResult = "1,2,12\n" +
+ "1,3,13\n" +
+ "2,3,23\n" +
+ "3,4,34\n" +
+ "3,5,35\n" +
+ "4,5,45\n" +
+ "5,1,51\n" ;
+
+ compareResultAsTuples(result, expectedResult);
+ }
+
+ @Test
public void testTriplets() throws Exception {
/*
* Test getTriplets()
[4/5] flink git commit: [FLINK-2271] [FLINK-1522] [gelly] changed
PageRank example to expect an unweighted edge list added PageRank and
MusicProfiles tests got rid of unused suppress warning annotations in Graph
and JaccardSimilarityMeasure changed GSAPa
Posted by va...@apache.org.
[FLINK-2271] [FLINK-1522] [gelly] changed PageRank example to expect an unweighted edge list
added PageRank and MusicProfiles tests
got rid of unused suppress warning annotations in Graph and JaccardSimilarityMeasure
changed GSAPageRank example to expect an unweighted edge list
This closes #865
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/fe5bbdbf
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/fe5bbdbf
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/fe5bbdbf
Branch: refs/heads/master
Commit: fe5bbdbfbdefee3c1b150c661ff3292769c030bb
Parents: cad8510
Author: vasia <va...@apache.org>
Authored: Wed Jun 24 18:09:13 2015 +0200
Committer: vasia <va...@apache.org>
Committed: Thu Jun 25 22:58:17 2015 +0200
----------------------------------------------------------------------
.../main/java/org/apache/flink/graph/Graph.java | 3 -
.../apache/flink/graph/example/GSAPageRank.java | 13 ++-
.../graph/example/JaccardSimilarityMeasure.java | 2 -
.../apache/flink/graph/example/PageRank.java | 16 +--
.../graph/example/utils/MusicProfilesData.java | 23 ++++-
.../flink/graph/example/utils/PageRankData.java | 42 ++++++++
.../flink/graph/test/GatherSumApplyITCase.java | 17 ----
.../graph/test/example/MusicProfilesITCase.java | 101 +++++++++++++++++++
.../graph/test/example/PageRankITCase.java | 78 ++++++++++++++
9 files changed, 260 insertions(+), 35 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/fe5bbdbf/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/Graph.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/Graph.java b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/Graph.java
index 5e13ce1..a7c75bc 100755
--- a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/Graph.java
+++ b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/Graph.java
@@ -1025,7 +1025,6 @@ public class Graph<K, VV, EV> {
* @param vertex the vertex to be added
* @return the new graph containing the existing vertices as well as the one just added
*/
- @SuppressWarnings("unchecked")
public Graph<K, VV, EV> addVertex(final Vertex<K, VV> vertex) {
List<Vertex<K, VV>> newVertex = new ArrayList<Vertex<K, VV>>();
newVertex.add(vertex);
@@ -1040,7 +1039,6 @@ public class Graph<K, VV, EV> {
* @param verticesToAdd the list of vertices to add
* @return the new graph containing the existing and newly added vertices
*/
- @SuppressWarnings("unchecked")
public Graph<K, VV, EV> addVertices(List<Vertex<K, VV>> verticesToAdd) {
// Add the vertices
DataSet<Vertex<K, VV>> newVertices = this.vertices.union(this.context.fromCollection(verticesToAdd)).distinct();
@@ -1074,7 +1072,6 @@ public class Graph<K, VV, EV> {
* @param newEdges the data set of edges to be added
* @return a new graph containing the existing edges plus the newly added edges.
*/
- @SuppressWarnings("unchecked")
public Graph<K, VV, EV> addEdges(List<Edge<K, EV>> newEdges) {
DataSet<Edge<K,EV>> newEdgesDataSet = this.context.fromCollection(newEdges);
http://git-wip-us.apache.org/repos/asf/flink/blob/fe5bbdbf/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/GSAPageRank.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/GSAPageRank.java b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/GSAPageRank.java
index 47f67b8..45d4555 100644
--- a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/GSAPageRank.java
+++ b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/GSAPageRank.java
@@ -31,14 +31,13 @@ import org.apache.flink.graph.gsa.ApplyFunction;
import org.apache.flink.graph.gsa.GatherFunction;
import org.apache.flink.graph.gsa.Neighbor;
import org.apache.flink.graph.gsa.SumFunction;
-import org.apache.flink.graph.utils.Tuple3ToEdgeMap;
import org.apache.flink.util.Collector;
/**
* This example implements a simple PageRank algorithm, using a gather-sum-apply iteration.
*
- * The edges input file is expected to contain one edge per line, with long IDs and double
- * values, in the following format:"<sourceVertexID>\t<targetVertexID>\t<edgeValue>".
+ * The edges input file is expected to contain one edge per line, with long IDs and no
+ * values, in the following format:"<sourceVertexID>\t<targetVertexID>".
*
* If no arguments are provided, the example runs with a random graph of 10 vertices
* and random edge weights.
@@ -187,8 +186,12 @@ public class GSAPageRank implements ProgramDescription {
return env.readCsvFile(edgeInputPath)
.fieldDelimiter("\t")
.lineDelimiter("\n")
- .types(Long.class, Long.class, Double.class)
- .map(new Tuple3ToEdgeMap<Long, Double>());
+ .types(Long.class, Long.class)
+ .map(new MapFunction<Tuple2<Long, Long>, Edge<Long, Double>>() {
+ public Edge<Long, Double> map(Tuple2<Long, Long> input) {
+ return new Edge<Long, Double>(input.f0, input.f1, 1.0);
+ }
+ }).withForwardedFields("f0; f1");
}
return env.generateSequence(1, numPages).flatMap(
http://git-wip-us.apache.org/repos/asf/flink/blob/fe5bbdbf/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/JaccardSimilarityMeasure.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/JaccardSimilarityMeasure.java b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/JaccardSimilarityMeasure.java
index be241ce..79de407 100644
--- a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/JaccardSimilarityMeasure.java
+++ b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/JaccardSimilarityMeasure.java
@@ -114,7 +114,6 @@ public class JaccardSimilarityMeasure implements ProgramDescription {
/**
* Each vertex will have a HashSet containing its neighbor ids as value.
*/
- @SuppressWarnings("serial")
private static final class GatherNeighbors implements ReduceNeighborsFunction<HashSet<Long>> {
@Override
@@ -136,7 +135,6 @@ public class JaccardSimilarityMeasure implements ProgramDescription {
*
* The Jaccard similarity coefficient is then, the intersection/union.
*/
- @SuppressWarnings("serial")
private static final class ComputeJaccard implements
MapFunction<Triplet<Long, HashSet<Long>, Double>, Edge<Long, Double>> {
http://git-wip-us.apache.org/repos/asf/flink/blob/fe5bbdbf/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/PageRank.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/PageRank.java b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/PageRank.java
index e4ad13f..10b4be4 100644
--- a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/PageRank.java
+++ b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/PageRank.java
@@ -28,17 +28,15 @@ import org.apache.flink.graph.Edge;
import org.apache.flink.graph.Graph;
import org.apache.flink.graph.Vertex;
import org.apache.flink.graph.library.PageRankAlgorithm;
-import org.apache.flink.graph.utils.Tuple3ToEdgeMap;
import org.apache.flink.util.Collector;
/**
* This example implements a simple PageRank algorithm, using a vertex-centric iteration.
*
- * The edges input file is expected to contain one edge per line, with long IDs and double
- * values, in the following format:"<sourceVertexID>\t<targetVertexID>\t<edgeValue>".
+ * The edges input file is expected to contain one edge per line, with long IDs and no
+ * values, in the following format:"<sourceVertexID>\t<targetVertexID>".
*
- * If no arguments are provided, the example runs with a random graph of 10 vertices
- * and random edge weights.
+ * If no arguments are provided, the example runs with a random graph of 10 vertices.
*
*/
public class PageRank implements ProgramDescription {
@@ -131,8 +129,12 @@ public class PageRank implements ProgramDescription {
return env.readCsvFile(edgeInputPath)
.fieldDelimiter("\t")
.lineDelimiter("\n")
- .types(Long.class, Long.class, Double.class)
- .map(new Tuple3ToEdgeMap<Long, Double>());
+ .types(Long.class, Long.class)
+ .map(new MapFunction<Tuple2<Long, Long>, Edge<Long, Double>>() {
+ public Edge<Long, Double> map(Tuple2<Long, Long> input) {
+ return new Edge<Long, Double>(input.f0, input.f1, 1.0);
+ }
+ }).withForwardedFields("f0; f1");
}
return env.generateSequence(1, numPages).flatMap(
http://git-wip-us.apache.org/repos/asf/flink/blob/fe5bbdbf/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/utils/MusicProfilesData.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/utils/MusicProfilesData.java b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/utils/MusicProfilesData.java
index 0a7162d..6b96372 100644
--- a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/utils/MusicProfilesData.java
+++ b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/utils/MusicProfilesData.java
@@ -79,5 +79,26 @@ public class MusicProfilesData {
errors.add("ERROR: <song_15 track_15> Black Trees");
return env.fromCollection(errors);
}
-}
+ public static final String USER_SONG_TRIPLETS = "user_1 song_1 100\n" + "user_1 song_5 200\n"
+ + "user_2 song_1 10\n" + "user_2 song_4 20\n"
+ + "user_3 song_2 3\n"
+ + "user_4 song_2 1\n" + "user_4 song_3 2\n"
+ + "user_5 song_3 30";
+
+ public static final String MISMATCHES = "ERROR: <song_5 track_8> Angie";
+
+ public static final String MAX_ITERATIONS = "2";
+
+ public static final String TOP_SONGS_RESULT = "user_1 song_1\n" +
+ "user_2 song_4\n" +
+ "user_3 song_2\n" +
+ "user_4 song_3\n" +
+ "user_5 song_3";
+
+ public static final String COMMUNITIES_RESULT = "user_1 1\n" +
+ "user_2 1\n" +
+ "user_3 3\n" +
+ "user_4 3\n" +
+ "user_5 4";
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/fe5bbdbf/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/utils/PageRankData.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/utils/PageRankData.java b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/utils/PageRankData.java
new file mode 100644
index 0000000..077572e
--- /dev/null
+++ b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/utils/PageRankData.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.graph.example.utils;
+
+public class PageRankData {
+
+ public static final String EDGES = "2 1\n" +
+ "5 2\n" +
+ "5 4\n" +
+ "4 3\n" +
+ "4 2\n" +
+ "1 4\n" +
+ "1 2\n" +
+ "1 3\n" +
+ "3 5\n";
+
+
+ public static final String RANKS_AFTER_3_ITERATIONS = "1 0.237\n" +
+ "2 0.248\n" +
+ "3 0.173\n" +
+ "4 0.175\n" +
+ "5 0.165\n";
+
+ private PageRankData() {}
+}
+
http://git-wip-us.apache.org/repos/asf/flink/blob/fe5bbdbf/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyITCase.java
index ca40323..a883fa0 100755
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyITCase.java
@@ -20,7 +20,6 @@ package org.apache.flink.graph.test;
import com.google.common.base.Charsets;
import com.google.common.io.Files;
-import org.apache.flink.graph.example.GSAPageRank;
import org.apache.flink.graph.example.GSAConnectedComponents;
import org.apache.flink.graph.example.GSASingleSourceShortestPaths;
import org.apache.flink.test.util.MultipleProgramsTestBase;
@@ -98,22 +97,6 @@ public class GatherSumApplyITCase extends MultipleProgramsTestBase {
}
// --------------------------------------------------------------------------------------------
- // Page Rank Test
- // --------------------------------------------------------------------------------------------
-
- @Test
- public void testPageRank() throws Exception {
- GSAPageRank.main(new String[]{edgesPath, resultPath, "16"});
- expectedResult = "1 7.47880014315678E21\n" +
- "2 1.6383884499619055E21\n" +
- "3 3.044048626469292E21\n" +
- "4 1.6896936994425786E21\n" +
- "5 4.214827876275491E21\n" +
- "6 1.0\n" +
- "7 8.157142857142858";
- }
-
- // --------------------------------------------------------------------------------------------
// Sample data
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/fe5bbdbf/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/example/MusicProfilesITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/example/MusicProfilesITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/example/MusicProfilesITCase.java
new file mode 100644
index 0000000..0410d41
--- /dev/null
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/example/MusicProfilesITCase.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.graph.test.example;
+
+import com.google.common.base.Charsets;
+import com.google.common.io.Files;
+
+import org.apache.flink.graph.example.MusicProfiles;
+import org.apache.flink.graph.example.utils.MusicProfilesData;
+import org.apache.flink.test.util.MultipleProgramsTestBase;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+
+@RunWith(Parameterized.class)
+public class MusicProfilesITCase extends MultipleProgramsTestBase {
+
+ private String tripletsPath;
+
+ private String mismatchesPath;
+
+ private String topSongsResultPath;
+
+ private String communitiesResultPath;
+
+ private String expectedTopSongs;
+
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ public MusicProfilesITCase(TestExecutionMode mode) {
+ super(mode);
+ }
+
+ @Before
+ public void before() throws Exception {
+ topSongsResultPath = tempFolder.newFile().toURI().toString();
+ communitiesResultPath = tempFolder.newFile().toURI().toString();
+
+ File tripletsFile = tempFolder.newFile();
+ Files.write(MusicProfilesData.USER_SONG_TRIPLETS, tripletsFile, Charsets.UTF_8);
+ tripletsPath = tripletsFile.toURI().toString();
+
+ File mismatchesFile = tempFolder.newFile();
+ Files.write(MusicProfilesData.MISMATCHES, mismatchesFile, Charsets.UTF_8);
+ mismatchesPath = mismatchesFile.toURI().toString();
+ }
+
+ @Test
+ public void testMusicProfilesExample() throws Exception {
+ MusicProfiles.main(new String[]{tripletsPath, mismatchesPath, topSongsResultPath, communitiesResultPath,
+ MusicProfilesData.MAX_ITERATIONS + ""});
+ expectedTopSongs = MusicProfilesData.TOP_SONGS_RESULT;
+ }
+
+ @After
+ public void after() throws Exception {
+ compareResultsByLinesInMemory(expectedTopSongs, topSongsResultPath);
+
+ ArrayList<String> list = new ArrayList<String>();
+ readAllResultLines(list, communitiesResultPath, new String[]{}, false);
+
+ String[] result = list.toArray(new String[list.size()]);
+ Arrays.sort(result);
+
+ // check that user_1 and user_2 are in the same community
+ Assert.assertEquals("users 1 and 2 are not in the same community",
+ result[0].substring(7), result[1].substring(7));
+
+ // check that user_3, user_4 and user_5 are in the same community
+ Assert.assertEquals("users 3 and 4 are not in the same community",
+ result[2].substring(7), result[3].substring(7));
+ Assert.assertEquals("users 4 and 5 are not in the same community",
+ result[3].substring(7), result[4].substring(7));
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/fe5bbdbf/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/example/PageRankITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/example/PageRankITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/example/PageRankITCase.java
new file mode 100644
index 0000000..544cc66
--- /dev/null
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/example/PageRankITCase.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.graph.test.example;
+
+import java.io.File;
+
+import com.google.common.base.Charsets;
+import com.google.common.io.Files;
+
+import org.apache.flink.graph.example.GSAPageRank;
+import org.apache.flink.graph.example.PageRank;
+import org.apache.flink.graph.example.utils.PageRankData;
+import org.apache.flink.test.util.MultipleProgramsTestBase;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class PageRankITCase extends MultipleProgramsTestBase {
+
+ public PageRankITCase(TestExecutionMode mode){
+ super(mode);
+ }
+
+ private String edgesPath;
+ private String resultPath;
+ private String expected;
+
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ @Before
+ public void before() throws Exception{
+ resultPath = tempFolder.newFile().toURI().toString();
+
+ File edgesFile = tempFolder.newFile();
+ Files.write(PageRankData.EDGES, edgesFile, Charsets.UTF_8);
+
+ edgesPath = edgesFile.toURI().toString();
+ }
+
+ @After
+ public void after() throws Exception{
+ compareKeyValueParisWithDelta(expected, resultPath, "\t", 0.01);
+ }
+
+ @Test
+ public void testPageRankWithThreeIterations() throws Exception {
+ PageRank.main(new String[] {edgesPath, resultPath, "3"});
+ expected = PageRankData.RANKS_AFTER_3_ITERATIONS;
+ }
+
+ @Test
+ public void testGSAPageRankWithThreeIterations() throws Exception {
+ GSAPageRank.main(new String[] {edgesPath, resultPath, "3"});
+ expected = PageRankData.RANKS_AFTER_3_ITERATIONS;
+ }
+}