You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2015/02/11 11:49:46 UTC
[44/50] [abbrv] flink git commit: [FLINK-1201] [gelly] use
MultipleProgramsTestBase for the tests
[FLINK-1201] [gelly] use MultipleProgramsTestBase for the tests
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e6b9cecd
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e6b9cecd
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e6b9cecd
Branch: refs/heads/master
Commit: e6b9cecd02b7606057ec589fe9eee8ea03825a80
Parents: 902df07
Author: vasia <va...@gmail.com>
Authored: Sun Jan 25 13:15:55 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Feb 11 10:46:16 2015 +0100
----------------------------------------------------------------------
.../apache/flink/graph/test/TestDegrees.java | 273 +++--
.../flink/graph/test/TestFromCollection.java | 182 ++--
.../flink/graph/test/TestGraphCreation.java | 252 ++---
.../graph/test/TestGraphCreationWithMapper.java | 232 ++--
.../flink/graph/test/TestGraphMutations.java | 467 ++++----
.../flink/graph/test/TestGraphOperations.java | 445 ++++----
.../flink/graph/test/TestJoinWithEdges.java | 1018 ++++++++----------
.../flink/graph/test/TestJoinWithVertices.java | 361 +++----
.../apache/flink/graph/test/TestMapEdges.java | 329 +++---
.../flink/graph/test/TestMapVertices.java | 363 ++++---
.../graph/test/TestReduceOnEdgesMethods.java | 527 ++++-----
.../graph/test/TestReduceOnNeighborMethods.java | 479 ++++----
.../flink/graph/test/TestWeaklyConnected.java | 183 ++--
13 files changed, 2440 insertions(+), 2671 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e6b9cecd/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestDegrees.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestDegrees.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestDegrees.java
index 8694c66..c572647 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestDegrees.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestDegrees.java
@@ -19,170 +19,153 @@
package org.apache.flink.graph.test;
import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.graph.Graph;
-import org.apache.flink.test.util.JavaProgramTestBase;
+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;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.util.Collection;
-import java.util.LinkedList;
-
@RunWith(Parameterized.class)
-public class TestDegrees extends JavaProgramTestBase {
+public class TestDegrees extends MultipleProgramsTestBase {
- private static int NUM_PROGRAMS = 6;
+ public TestDegrees(MultipleProgramsTestBase.ExecutionMode mode){
+ super(mode);
+ }
- private int curProgId = config.getInteger("ProgramId", -1);
private String resultPath;
private String expectedResult;
- public TestDegrees(Configuration config) {
- super(config);
+ @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 {
+ /*
+ * Test outDegrees()
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+
+ graph.outDegrees().writeAsCsv(resultPath);
+ env.execute();
+
+ expectedResult = "1,2\n" +
+ "2,1\n" +
+ "3,2\n" +
+ "4,1\n" +
+ "5,1\n";
}
- @Override
- protected void preSubmit() throws Exception {
- resultPath = getTempDirPath("result");
- }
+ @Test
+ public void testOutDegreesWithNoOutEdges() throws Exception {
+ /*
+ * Test outDegrees() no outgoing edges
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- @Override
- protected void testProgram() throws Exception {
- expectedResult = GraphProgs.runProgram(curProgId, resultPath);
- }
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeDataWithZeroDegree(env), env);
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(expectedResult, resultPath);
- }
+ graph.outDegrees().writeAsCsv(resultPath);
+ env.execute();
- @Parameterized.Parameters
- public static Collection<Object[]> getConfigurations() throws FileNotFoundException, IOException {
+ expectedResult = "1,3\n" +
+ "2,1\n" +
+ "3,1\n" +
+ "4,1\n" +
+ "5,0\n";
+ }
- LinkedList<Configuration> tConfigs = new LinkedList<Configuration>();
+ @Test
+ public void testInDegrees() throws Exception {
+ /*
+ * Test inDegrees()
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+
+ graph.inDegrees().writeAsCsv(resultPath);
+ env.execute();
+ expectedResult = "1,1\n" +
+ "2,1\n" +
+ "3,2\n" +
+ "4,1\n" +
+ "5,2\n";
+ }
- for(int i=1; i <= NUM_PROGRAMS; i++) {
- Configuration config = new Configuration();
- config.setInteger("ProgramId", i);
- tConfigs.add(config);
- }
+ @Test
+ public void testInDegreesWithNoInEdge() throws Exception {
+ /*
+ * Test inDegrees() no ingoing edge
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeDataWithZeroDegree(env), env);
+
+ graph.inDegrees().writeAsCsv(resultPath);
+ env.execute();
+ expectedResult = "1,0\n" +
+ "2,1\n" +
+ "3,1\n" +
+ "4,1\n" +
+ "5,3\n";
+ }
- return toParameterList(tConfigs);
+ @Test
+ public void testGetDegrees() throws Exception {
+ /*
+ * Test getDegrees()
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+
+ graph.getDegrees().writeAsCsv(resultPath);
+ env.execute();
+ expectedResult = "1,3\n" +
+ "2,2\n" +
+ "3,4\n" +
+ "4,2\n" +
+ "5,3\n";
}
- private static class GraphProgs {
-
- public static String runProgram(int progId, String resultPath) throws Exception {
-
- switch (progId) {
- case 1: {
- /*
- * Test outDegrees()
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- TestGraphUtils.getLongLongEdgeData(env), env);
-
- graph.outDegrees().writeAsCsv(resultPath);
- env.execute();
- return "1,2\n" +
- "2,1\n" +
- "3,2\n" +
- "4,1\n" +
- "5,1\n";
- }
- case 2: {
- /*
- * Test outDegrees() no outgoing edges
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- TestGraphUtils.getLongLongEdgeDataWithZeroDegree(env), env);
-
- graph.outDegrees().writeAsCsv(resultPath);
- env.execute();
- return "1,3\n" +
- "2,1\n" +
- "3,1\n" +
- "4,1\n" +
- "5,0\n";
- }
- case 3: {
- /*
- * Test inDegrees()
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- TestGraphUtils.getLongLongEdgeData(env), env);
-
- graph.inDegrees().writeAsCsv(resultPath);
- env.execute();
- return "1,1\n" +
- "2,1\n" +
- "3,2\n" +
- "4,1\n" +
- "5,2\n";
- }
- case 4: {
- /*
- * Test inDegrees() no ingoing edge
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- TestGraphUtils.getLongLongEdgeDataWithZeroDegree(env), env);
-
- graph.inDegrees().writeAsCsv(resultPath);
- env.execute();
- return "1,0\n" +
- "2,1\n" +
- "3,1\n" +
- "4,1\n" +
- "5,3\n";
- }
- case 5: {
- /*
- * Test getDegrees()
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- TestGraphUtils.getLongLongEdgeData(env), env);
-
- graph.getDegrees().writeAsCsv(resultPath);
- env.execute();
- return "1,3\n" +
- "2,2\n" +
- "3,4\n" +
- "4,2\n" +
- "5,3\n";
- }
- case 6: {
- /*
- * Test getDegrees() with disconnected data
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, NullValue, Long> graph =
- Graph.fromDataSet(TestGraphUtils.getDisconnectedLongLongEdgeData(env), env);
-
- graph.outDegrees().writeAsCsv(resultPath);
- env.execute();
- return "1,2\n" +
- "2,1\n" +
- "3,0\n" +
- "4,1\n" +
- "5,0\n";
- }
- default:
- throw new IllegalArgumentException("Invalid program id");
- }
- }
+ @Test
+ public void testGetDegreesWithDisconnectedData() throws Exception {
+ /*
+ * Test getDegrees() with disconnected data
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, NullValue, Long> graph =
+ Graph.fromDataSet(TestGraphUtils.getDisconnectedLongLongEdgeData(env), env);
+
+ graph.outDegrees().writeAsCsv(resultPath);
+ env.execute();
+ expectedResult = "1,2\n" +
+ "2,1\n" +
+ "3,0\n" +
+ "4,1\n" +
+ "5,0\n";
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/e6b9cecd/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestFromCollection.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestFromCollection.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestFromCollection.java
index 063e58b..e1b96e8 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestFromCollection.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestFromCollection.java
@@ -20,123 +20,101 @@ package org.apache.flink.graph.test;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.graph.Graph;
-import org.apache.flink.test.util.JavaProgramTestBase;
+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;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.util.Collection;
-import java.util.LinkedList;
-
@RunWith(Parameterized.class)
-public class TestFromCollection extends JavaProgramTestBase {
- private static int NUM_PROGRAMS = 3;
+public class TestFromCollection extends MultipleProgramsTestBase {
+
+ public TestFromCollection(MultipleProgramsTestBase.ExecutionMode mode){
+ super(mode);
+ }
- private int curProgId = config.getInteger("ProgramId", -1);
private String resultPath;
private String expectedResult;
- public TestFromCollection(Configuration config) {
- super(config);
- }
-
- @Override
- protected void preSubmit() throws Exception {
- resultPath = getTempDirPath("result");
+ @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 {
+ /*
+ * Test fromCollection(vertices, edges):
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ Graph<Long, Long, Long> graph = Graph.fromCollection(TestGraphUtils.getLongLongVertices(),
+ TestGraphUtils.getLongLongEdges(), env);
+
+ graph.getEdges().writeAsCsv(resultPath);
+ env.execute();
+ 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";
}
- @Override
- protected void testProgram() throws Exception {
- expectedResult = GraphProgs.runProgram(curProgId, resultPath);
+ @Test
+ public void testFromCollectionEdgesNoInitialValue() throws Exception {
+ /*
+ * Test fromCollection(edges) with no initial value for the vertices
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ Graph<Long, NullValue, Long> graph = Graph.fromCollection(TestGraphUtils.getLongLongEdges(),
+ env);
+
+ graph.getVertices().writeAsCsv(resultPath);
+ env.execute();
+ expectedResult = "1,(null)\n" +
+ "2,(null)\n" +
+ "3,(null)\n" +
+ "4,(null)\n" +
+ "5,(null)\n";
}
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(expectedResult, resultPath);
- }
-
- @Parameterized.Parameters
- public static Collection<Object[]> getConfigurations() throws FileNotFoundException, IOException {
-
- LinkedList<Configuration> tConfigs = new LinkedList<Configuration>();
-
- for(int i=1; i <= NUM_PROGRAMS; i++) {
- Configuration config = new Configuration();
- config.setInteger("ProgramId", i);
- tConfigs.add(config);
- }
-
- return toParameterList(tConfigs);
+ @Test
+ public void testFromCollectionEdgesWithInitialValue() throws Exception {
+ /*
+ * Test fromCollection(edges) with vertices initialised by a
+ * function that takes the id and doubles it
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ Graph<Long, Long, Long> graph = Graph.fromCollection(TestGraphUtils.getLongLongEdges(),
+ new InitVerticesMapper(), env);
+
+ graph.getVertices().writeAsCsv(resultPath);
+ env.execute();
+ expectedResult = "1,2\n" +
+ "2,4\n" +
+ "3,6\n" +
+ "4,8\n" +
+ "5,10\n";
}
- private static class GraphProgs {
-
- @SuppressWarnings("serial")
- public static String runProgram(int progId, String resultPath) throws Exception {
-
- switch (progId) {
- case 1: {
- /*
- * Test fromCollection(vertices, edges):
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- Graph<Long, Long, Long> graph = Graph.fromCollection(TestGraphUtils.getLongLongVertices(),
- TestGraphUtils.getLongLongEdges(), env);
-
- graph.getEdges().writeAsCsv(resultPath);
- env.execute();
- return "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";
- }
- case 2: {
- /*
- * Test fromCollection(edges) with no initial value for the vertices
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- Graph<Long, NullValue, Long> graph = Graph.fromCollection(TestGraphUtils.getLongLongEdges(),
- env);
-
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
- return "1,(null)\n" +
- "2,(null)\n" +
- "3,(null)\n" +
- "4,(null)\n" +
- "5,(null)\n";
- }
- case 3: {
- /*
- * Test fromCollection(edges) with vertices initialised by a
- * function that takes the id and doubles it
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- Graph<Long, Long, Long> graph = Graph.fromCollection(TestGraphUtils.getLongLongEdges(),
- new MapFunction<Long, Long>() {
- public Long map(Long vertexId) {
- return vertexId * 2;
- }
- }, env);
-
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
- return "1,2\n" +
- "2,4\n" +
- "3,6\n" +
- "4,8\n" +
- "5,10\n";
- }
- default:
- throw new IllegalArgumentException("Invalid program id");
- }
+ @SuppressWarnings("serial")
+ private static final class InitVerticesMapper implements MapFunction<Long, Long> {
+ public Long map(Long vertexId) {
+ return vertexId * 2;
}
- }
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/e6b9cecd/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphCreation.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphCreation.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphCreation.java
index 241a0f5..d3e3bda 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphCreation.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphCreation.java
@@ -18,169 +18,153 @@
package org.apache.flink.graph.test;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.util.Collection;
-import java.util.LinkedList;
-
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.configuration.Configuration;
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.DummyCustomParameterizedType;
import org.apache.flink.graph.validation.InvalidVertexIdsValidator;
-import org.apache.flink.test.util.JavaProgramTestBase;
+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;
-import org.junit.runners.Parameterized.Parameters;
@RunWith(Parameterized.class)
-public class TestGraphCreation extends JavaProgramTestBase {
+public class TestGraphCreation extends MultipleProgramsTestBase {
+
+ public TestGraphCreation(MultipleProgramsTestBase.ExecutionMode mode){
+ super(mode);
+ }
- private static int NUM_PROGRAMS = 5;
+ private String resultPath;
+ private String expectedResult;
- private int curProgId = config.getInteger("ProgramId", -1);
- private String resultPath;
- private String expectedResult;
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
- public TestGraphCreation(Configuration config) {
- super(config);
+ @Before
+ public void before() throws Exception{
+ resultPath = tempFolder.newFile().toURI().toString();
}
- @Override
- protected void preSubmit() throws Exception {
- resultPath = getTempDirPath("result");
+ @After
+ public void after() throws Exception{
+ compareResultsByLinesInMemory(expectedResult, resultPath);
}
- @Override
- protected void testProgram() throws Exception {
- expectedResult = GraphProgs.runProgram(curProgId, resultPath);
+ @Test
+ public void testCreateWithoutVertexValues() throws Exception {
+ /*
+ * Test create() with edge dataset and no vertex values
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ Graph<Long, NullValue, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongEdgeData(env), env);
+
+ graph.getVertices().writeAsCsv(resultPath);
+ env.execute();
+ expectedResult = "1,(null)\n" +
+ "2,(null)\n" +
+ "3,(null)\n" +
+ "4,(null)\n" +
+ "5,(null)\n";
}
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(expectedResult, resultPath);
+ @Test
+ public void testCreateWithMapper() throws Exception {
+ /*
+ * Test create() with edge dataset and a mapper that assigns the id as value
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongEdgeData(env),
+ new AssignIdAsValueMapper(), env);
+
+ graph.getVertices().writeAsCsv(resultPath);
+ env.execute();
+ expectedResult = "1,1\n" +
+ "2,2\n" +
+ "3,3\n" +
+ "4,4\n" +
+ "5,5\n";
}
- @Parameters
- public static Collection<Object[]> getConfigurations() throws FileNotFoundException, IOException {
+ @Test
+ public void testCreateWithCustomVertexValue() throws Exception {
+ /*
+ * Test create() with edge dataset and a mapper that assigns a parametrized custom vertex value
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ Graph<Long, DummyCustomParameterizedType<Double>, Long> graph = Graph.fromDataSet(
+ TestGraphUtils.getLongLongEdgeData(env), new AssignCustomVertexValueMapper(), env);
+
+ graph.getVertices().writeAsCsv(resultPath);
+ env.execute();
+ 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";
+ }
- LinkedList<Configuration> tConfigs = new LinkedList<Configuration>();
+ @Test
+ public void testValidate() throws Exception {
+ /*
+ * Test validate():
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ DataSet<Vertex<Long, Long>> vertices = TestGraphUtils.getLongLongVertexData(env);
+ DataSet<Edge<Long, Long>> edges = TestGraphUtils.getLongLongEdgeData(env);
- for(int i=1; i <= NUM_PROGRAMS; i++) {
- Configuration config = new Configuration();
- config.setInteger("ProgramId", i);
- tConfigs.add(config);
- }
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(vertices, edges, env);
+ DataSet<Boolean> result = graph.validate(new InvalidVertexIdsValidator<Long, Long, Long>());
- return toParameterList(tConfigs);
+ result.writeAsText(resultPath);
+ env.execute();
+
+ expectedResult = "true\n";
+ }
+
+ @Test
+ public void testValidateWithInvalidIds() throws Exception {
+ /*
+ * Test validate() - invalid vertex ids
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ DataSet<Vertex<Long, Long>> vertices = TestGraphUtils.getLongLongInvalidVertexData(env);
+ DataSet<Edge<Long, Long>> edges = TestGraphUtils.getLongLongEdgeData(env);
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(vertices, edges, env);
+ DataSet<Boolean> result = graph.validate(new InvalidVertexIdsValidator<Long, Long, Long>());
+ result.writeAsText(resultPath);
+ env.execute();
+
+ expectedResult = "false\n";
}
- private static class GraphProgs {
-
- @SuppressWarnings("serial")
- public static String runProgram(int progId, String resultPath) throws Exception {
-
- switch(progId) {
- case 1: {
- /*
- * Test create() with edge dataset and no vertex values
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- Graph<Long, NullValue, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongEdgeData(env), env);
-
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
- return "1,(null)\n" +
- "2,(null)\n" +
- "3,(null)\n" +
- "4,(null)\n" +
- "5,(null)\n";
- }
- case 2: {
- /*
- * Test create() with edge dataset and a mapper that assigns the id as value
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongEdgeData(env),
- new MapFunction<Long, Long>() {
- public Long map(Long vertexId) {
- return vertexId;
- }
- }, env);
-
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
- return "1,1\n" +
- "2,2\n" +
- "3,3\n" +
- "4,4\n" +
- "5,5\n";
- }
- case 3: {
- /*
- * Test create() with edge dataset and a mapper that assigns a parametrized custom vertex value
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- Graph<Long, DummyCustomParameterizedType<Double>, Long> graph = Graph.fromDataSet(
- TestGraphUtils.getLongLongEdgeData(env),
- new MapFunction<Long, DummyCustomParameterizedType<Double>>() {
-
- DummyCustomParameterizedType<Double> dummyValue =
- new DummyCustomParameterizedType<Double>();
-
- public DummyCustomParameterizedType<Double> map(Long vertexId) {
- dummyValue.setIntField(vertexId.intValue()-1);
- dummyValue.setTField(vertexId*2.0);
- return dummyValue;
- }
- }, env);
-
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
- return "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";
- }
- case 4: {
- /*
- * Test validate():
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- DataSet<Vertex<Long, Long>> vertices = TestGraphUtils.getLongLongVertexData(env);
- DataSet<Edge<Long, Long>> edges = TestGraphUtils.getLongLongEdgeData(env);
- Graph<Long, Long, Long> graph = Graph.fromDataSet(vertices, edges, env);
- DataSet<Boolean> result = graph.validate(new InvalidVertexIdsValidator<Long, Long, Long>());
- result.writeAsText(resultPath);
- env.execute();
-
- return "true\n";
- }
- case 5: {
- /*
- * Test validate() - invalid vertex ids
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- DataSet<Vertex<Long, Long>> vertices = TestGraphUtils.getLongLongInvalidVertexData(env);
- DataSet<Edge<Long, Long>> edges = TestGraphUtils.getLongLongEdgeData(env);
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(vertices, edges, env);
- DataSet<Boolean> result = graph.validate(new InvalidVertexIdsValidator<Long, Long, Long>());
- result.writeAsText(resultPath);
- env.execute();
-
- return "false\n";
- }
- default:
- throw new IllegalArgumentException("Invalid program id");
- }
- }
+ @SuppressWarnings("serial")
+ private static final class AssignIdAsValueMapper implements MapFunction<Long, Long> {
+ public Long map(Long vertexId) {
+ return vertexId;
}
+ }
+ @SuppressWarnings("serial")
+ private static final class AssignCustomVertexValueMapper implements
+ MapFunction<Long, DummyCustomParameterizedType<Double>> {
+
+ DummyCustomParameterizedType<Double> dummyValue =
+ new DummyCustomParameterizedType<Double>();
+
+ public DummyCustomParameterizedType<Double> map(Long vertexId) {
+ dummyValue.setIntField(vertexId.intValue()-1);
+ dummyValue.setTField(vertexId*2.0);
+ return dummyValue;
+ }
}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/e6b9cecd/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphCreationWithMapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphCreationWithMapper.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphCreationWithMapper.java
index edf9e26..67ff5cc 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphCreationWithMapper.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphCreationWithMapper.java
@@ -18,155 +18,141 @@
package org.apache.flink.graph.test;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.util.Collection;
-import java.util.LinkedList;
-
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.graph.Graph;
import org.apache.flink.graph.test.TestGraphUtils.DummyCustomType;
-import org.apache.flink.test.util.JavaProgramTestBase;
+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;
-import org.junit.runners.Parameterized.Parameters;
@RunWith(Parameterized.class)
-public class TestGraphCreationWithMapper extends JavaProgramTestBase {
+public class TestGraphCreationWithMapper extends MultipleProgramsTestBase {
+
+ public TestGraphCreationWithMapper(MultipleProgramsTestBase.ExecutionMode mode){
+ super(mode);
+ }
- private static int NUM_PROGRAMS = 4;
+ private String resultPath;
+ private String expectedResult;
- private int curProgId = config.getInteger("ProgramId", -1);
- private String resultPath;
- private String expectedResult;
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
- public TestGraphCreationWithMapper(Configuration config) {
- super(config);
+ @Before
+ public void before() throws Exception{
+ resultPath = tempFolder.newFile().toURI().toString();
}
- @Override
- protected void preSubmit() throws Exception {
- resultPath = getTempDirPath("result");
+ @After
+ public void after() throws Exception{
+ compareResultsByLinesInMemory(expectedResult, resultPath);
}
- @Override
- protected void testProgram() throws Exception {
- expectedResult = GraphProgs.runProgram(curProgId, resultPath);
+ @Test
+ public void testWithDoubleValueMapper() throws Exception {
+ /*
+ * Test create() with edge dataset and a mapper that assigns a double constant as value
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ Graph<Long, Double, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongEdgeData(env),
+ new AssignDoubleValueMapper(), env);
+
+ graph.getVertices().writeAsCsv(resultPath);
+ env.execute();
+ expectedResult = "1,0.1\n" +
+ "2,0.1\n" +
+ "3,0.1\n" +
+ "4,0.1\n" +
+ "5,0.1\n";
}
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(expectedResult, resultPath);
+ @Test
+ public void testWithTuple2ValueMapper() throws Exception {
+ /*
+ * Test create() with edge dataset and a mapper that assigns a Tuple2 as value
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ Graph<Long, Tuple2<Long, Long>, Long> graph = Graph.fromDataSet(
+ TestGraphUtils.getLongLongEdgeData(env), new AssignTuple2ValueMapper(), env);
+
+ graph.getVertices().writeAsCsv(resultPath);
+ env.execute();
+ expectedResult = "1,(2,42)\n" +
+ "2,(4,42)\n" +
+ "3,(6,42)\n" +
+ "4,(8,42)\n" +
+ "5,(10,42)\n";
}
- @Parameters
- public static Collection<Object[]> getConfigurations() throws FileNotFoundException, IOException {
+ @Test
+ public void testWithConstantValueMapper() throws Exception {
+ /*
+ * Test create() with edge dataset with String key type
+ * and a mapper that assigns a double constant as value
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ Graph<String, Double, Long> graph = Graph.fromDataSet(TestGraphUtils.getStringLongEdgeData(env),
+ new AssignDoubleConstantMapper(), env);
+
+ graph.getVertices().writeAsCsv(resultPath);
+ env.execute();
+ expectedResult = "1,0.1\n" +
+ "2,0.1\n" +
+ "3,0.1\n" +
+ "4,0.1\n" +
+ "5,0.1\n";
+ }
- LinkedList<Configuration> tConfigs = new LinkedList<Configuration>();
+ @Test
+ public void testWithDCustomValueMapper() throws Exception {
+ /*
+ * Test create() with edge dataset and a mapper that assigns a custom vertex value
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ Graph<Long, DummyCustomType, Long> graph = Graph.fromDataSet(
+ TestGraphUtils.getLongLongEdgeData(env), new AssignCustomValueMapper(), env);
+
+ graph.getVertices().writeAsCsv(resultPath);
+ env.execute();
+ expectedResult = "1,(F,0)\n" +
+ "2,(F,1)\n" +
+ "3,(F,2)\n" +
+ "4,(F,3)\n" +
+ "5,(F,4)\n";
+ }
- for(int i=1; i <= NUM_PROGRAMS; i++) {
- Configuration config = new Configuration();
- config.setInteger("ProgramId", i);
- tConfigs.add(config);
+ @SuppressWarnings("serial")
+ private static final class AssignDoubleValueMapper implements MapFunction<Long, Double> {
+ public Double map(Long value) {
+ return 0.1d;
}
+ }
- return toParameterList(tConfigs);
+ @SuppressWarnings("serial")
+ private static final class AssignTuple2ValueMapper implements MapFunction<Long, Tuple2<Long, Long>> {
+ public Tuple2<Long, Long> map(Long vertexId) {
+ return new Tuple2<Long, Long>(vertexId*2, 42l);
+ }
}
- private static class GraphProgs {
-
- @SuppressWarnings("serial")
- public static String runProgram(int progId, String resultPath) throws Exception {
-
- switch(progId) {
- case 1: {
- /*
- * Test create() with edge dataset and a mapper that assigns a double constant as value
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- Graph<Long, Double, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongEdgeData(env),
- new MapFunction<Long, Double>() {
- public Double map(Long value) {
- return 0.1d;
- }
- }, env);
-
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
- return "1,0.1\n" +
- "2,0.1\n" +
- "3,0.1\n" +
- "4,0.1\n" +
- "5,0.1\n";
- }
- case 2: {
- /*
- * Test create() with edge dataset and a mapper that assigns a Tuple2 as value
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- Graph<Long, Tuple2<Long, Long>, Long> graph = Graph.fromDataSet(
- TestGraphUtils.getLongLongEdgeData(env), new MapFunction<Long, Tuple2<Long, Long>>() {
- public Tuple2<Long, Long> map(Long vertexId) {
- return new Tuple2<Long, Long>(vertexId*2, 42l);
- }
- }, env);
-
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
- return "1,(2,42)\n" +
- "2,(4,42)\n" +
- "3,(6,42)\n" +
- "4,(8,42)\n" +
- "5,(10,42)\n";
- }
- case 3: {
- /*
- * Test create() with edge dataset with String key type
- * and a mapper that assigns a double constant as value
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- Graph<String, Double, Long> graph = Graph.fromDataSet(TestGraphUtils.getStringLongEdgeData(env),
- new MapFunction<String, Double>() {
- public Double map(String value) {
- return 0.1d;
- }
- }, env);
-
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
- return "1,0.1\n" +
- "2,0.1\n" +
- "3,0.1\n" +
- "4,0.1\n" +
- "5,0.1\n";
- }
- case 4: {
- /*
- * Test create() with edge dataset and a mapper that assigns a custom vertex value
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- Graph<Long, DummyCustomType, Long> graph = Graph.fromDataSet(
- TestGraphUtils.getLongLongEdgeData(env), new MapFunction<Long, DummyCustomType>() {
- public DummyCustomType map(Long vertexId) {
- return new DummyCustomType(vertexId.intValue()-1, false);
- }
- }, env);
-
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
- return "1,(F,0)\n" +
- "2,(F,1)\n" +
- "3,(F,2)\n" +
- "4,(F,3)\n" +
- "5,(F,4)\n";
- }
- default:
- throw new IllegalArgumentException("Invalid program id");
- }
- }
+ @SuppressWarnings("serial")
+ private static final class AssignDoubleConstantMapper implements MapFunction<String, Double> {
+ public Double map(String value) {
+ return 0.1d;
}
+ }
+ @SuppressWarnings("serial")
+ private static final class AssignCustomValueMapper implements MapFunction<Long, DummyCustomType> {
+ public DummyCustomType map(Long vertexId) {
+ return new DummyCustomType(vertexId.intValue()-1, false);
+ }
}
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/e6b9cecd/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphMutations.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphMutations.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphMutations.java
index b8f1af2..f53f51e 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphMutations.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphMutations.java
@@ -18,113 +18,89 @@
package org.apache.flink.graph.test;
-import java.io.FileNotFoundException;
-import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collection;
-import java.util.LinkedList;
import java.util.List;
import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.graph.Edge;
import org.apache.flink.graph.Graph;
import org.apache.flink.graph.Vertex;
-import org.apache.flink.test.util.JavaProgramTestBase;
+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;
-import org.junit.runners.Parameterized.Parameters;
@RunWith(Parameterized.class)
-public class TestGraphMutations extends JavaProgramTestBase {
+public class TestGraphMutations extends MultipleProgramsTestBase {
- private static int NUM_PROGRAMS = 9;
-
- private int curProgId = config.getInteger("ProgramId", -1);
- private String resultPath;
- private String expectedResult;
-
- public TestGraphMutations(Configuration config) {
- super(config);
- }
-
- @Override
- protected void preSubmit() throws Exception {
- resultPath = getTempDirPath("result");
+ public TestGraphMutations(MultipleProgramsTestBase.ExecutionMode mode){
+ super(mode);
}
- @Override
- protected void testProgram() throws Exception {
- expectedResult = GraphProgs.runProgram(curProgId, resultPath);
+ private String resultPath;
+ private String expectedResult;
+
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ @Before
+ public void before() throws Exception{
+ resultPath = tempFolder.newFile().toURI().toString();
}
-
- @Override
- protected void postSubmit() throws Exception {
+
+ @After
+ public void after() throws Exception{
compareResultsByLinesInMemory(expectedResult, resultPath);
}
-
- @Parameters
- public static Collection<Object[]> getConfigurations() throws FileNotFoundException, IOException {
- LinkedList<Configuration> tConfigs = new LinkedList<Configuration>();
+ @Test
+ public void testAddVertex() throws Exception {
+ /*
+ * Test addVertex() -- simple case
+ */
- for(int i=1; i <= NUM_PROGRAMS; i++) {
- Configuration config = new Configuration();
- config.setInteger("ProgramId", i);
- tConfigs.add(config);
- }
-
- return toParameterList(tConfigs);
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+
+ List<Edge<Long, Long>> edges = new ArrayList<Edge<Long, Long>>();
+ edges.add(new Edge<Long, Long>(6L, 1L, 61L));
+ graph = graph.addVertex(new Vertex<Long, Long>(6L, 6L), edges);
+ graph.getEdges().writeAsCsv(resultPath);
+ env.execute();
+
+ 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" +
+ "6,1,61\n";
}
-
- private static class GraphProgs {
+
+ @Test
+ public void testAddVertexExisting() throws Exception {
+ /*
+ * Test addVertex() -- add an existing vertex
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- public static String runProgram(int progId, String resultPath) throws Exception {
-
- switch(progId) {
- case 1: {
- /*
- * Test addVertex() -- simple case
- */
-
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- TestGraphUtils.getLongLongEdgeData(env), env);
-
- List<Edge<Long, Long>> edges = new ArrayList<Edge<Long, Long>>();
- edges.add(new Edge<Long, Long>(6L, 1L, 61L));
- graph = graph.addVertex(new Vertex<Long, Long>(6L, 6L), edges);
- graph.getEdges().writeAsCsv(resultPath);
- env.execute();
-
- return "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" +
- "6,1,61\n";
-
- }
- case 2: {
- /*
- * Test addVertex() -- add an existing vertex
- */
-
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- TestGraphUtils.getLongLongEdgeData(env), env);
-
- List<Edge<Long, Long>> edges = new ArrayList<Edge<Long, Long>>();
- edges.add(new Edge<Long, Long>(1L, 5L, 15L));
- graph = graph.addVertex(new Vertex<Long, Long>(1L, 1L), edges);
- graph.getEdges().writeAsCsv(resultPath);
- env.execute();
-
- return "1,2,12\n" +
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+
+ List<Edge<Long, Long>> edges = new ArrayList<Edge<Long, Long>>();
+ edges.add(new Edge<Long, Long>(1L, 5L, 15L));
+ graph = graph.addVertex(new Vertex<Long, Long>(1L, 1L), edges);
+ graph.getEdges().writeAsCsv(resultPath);
+ env.execute();
+
+ expectedResult = "1,2,12\n" +
"1,3,13\n" +
"1,5,15\n" +
"2,3,23\n" +
@@ -132,163 +108,166 @@ public class TestGraphMutations extends JavaProgramTestBase {
"3,5,35\n" +
"4,5,45\n" +
"5,1,51\n";
-
- }
- case 3: {
- /*
- * Test addVertex() -- add vertex with empty edge set
- */
-
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- TestGraphUtils.getLongLongEdgeData(env), env);
- List<Edge<Long, Long>> edges = new ArrayList<Edge<Long, Long>>();
- graph = graph.addVertex(new Vertex<Long, Long>(6L, 6L), edges);
- graph.getVertices().writeAsCsv(resultPath);
- env.execute();
-
- return "1,1\n" +
- "2,2\n" +
- "3,3\n" +
- "4,4\n" +
- "5,5\n" +
- "6,6\n";
-
- }
- case 4: {
- /*
- * Test removeVertex() -- simple case
- */
-
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- 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();
-
- return "1,2,12\n" +
- "1,3,13\n" +
- "2,3,23\n" +
- "3,4,34\n";
-
- }
- case 5: {
- /*
- * Test removeVertex() -- remove an invalid vertex
- */
-
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- 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();
-
- return "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";
- }
- case 6: {
- /*
- * Test addEdge() -- simple case
- */
-
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- 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();
-
- return "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" +
- "6,1,61\n";
- }
- case 7: {
- /*
- * Test addEdge() -- add already existing edge
- */
-
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- 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();
-
- return "1,2,12\n" +
- "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";
- }
- case 8: {
- /*
- * Test removeEdge() -- simple case
- */
-
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- 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();
-
- return "1,2,12\n" +
- "1,3,13\n" +
- "2,3,23\n" +
- "3,4,34\n" +
- "3,5,35\n" +
- "4,5,45\n";
-
- }
- case 9: {
- /*
- * Test removeEdge() -- invalid edge
- */
-
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- 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();
-
- return "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";
-
- }
- default:
- throw new IllegalArgumentException("Invalid program id");
- }
- }
+ }
+
+ @Test
+ public void testAddVertexNoEdges() throws Exception {
+ /*
+ * Test addVertex() -- add vertex with empty edge set
+ */
+
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+ List<Edge<Long, Long>> edges = new ArrayList<Edge<Long, Long>>();
+ graph = graph.addVertex(new Vertex<Long, Long>(6L, 6L), edges);
+ graph.getVertices().writeAsCsv(resultPath);
+ env.execute();
+
+ expectedResult = "1,1\n" +
+ "2,2\n" +
+ "3,3\n" +
+ "4,4\n" +
+ "5,5\n" +
+ "6,6\n";
+ }
+
+ @Test
+ public void testRemoveVertex() throws Exception {
+ /*
+ * Test removeVertex() -- simple case
+ */
+
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ 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();
+
+ expectedResult = "1,2,12\n" +
+ "1,3,13\n" +
+ "2,3,23\n" +
+ "3,4,34\n";
+ }
+
+ @Test
+ public void testRemoveInvalidVertex() throws Exception {
+ /*
+ * Test removeVertex() -- remove an invalid vertex
+ */
+
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ 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();
+
+ 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";
}
-}
+ @Test
+ public void testAddEdge() throws Exception {
+ /*
+ * Test addEdge() -- simple case
+ */
+
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ 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();
+
+ 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" +
+ "6,1,61\n";
+ }
+
+ @Test
+ public void testAddExistingEdge() throws Exception {
+ /*
+ * Test addEdge() -- add already existing edge
+ */
+
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ 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();
+
+ expectedResult = "1,2,12\n" +
+ "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";
+ }
+
+ @Test
+ public void testRemoveVEdge() throws Exception {
+ /*
+ * Test removeEdge() -- simple case
+ */
+
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ 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();
+
+ 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";
+ }
+
+ @Test
+ public void testRemoveInvalidEdge() throws Exception {
+ /*
+ * Test removeEdge() -- invalid edge
+ */
+
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ 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();
+
+ 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";
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/e6b9cecd/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphOperations.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphOperations.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphOperations.java
index 9482997..6ab6928 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphOperations.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestGraphOperations.java
@@ -18,261 +18,250 @@
package org.apache.flink.graph.test;
-import java.io.FileNotFoundException;
-import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collection;
-import java.util.LinkedList;
import java.util.List;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.graph.Edge;
import org.apache.flink.graph.Graph;
import org.apache.flink.graph.Vertex;
-import org.apache.flink.test.util.JavaProgramTestBase;
+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;
-import org.junit.runners.Parameterized.Parameters;
@RunWith(Parameterized.class)
-public class TestGraphOperations extends JavaProgramTestBase {
+public class TestGraphOperations extends MultipleProgramsTestBase {
- private static int NUM_PROGRAMS = 10;
+ public TestGraphOperations(MultipleProgramsTestBase.ExecutionMode mode){
+ super(mode);
+ }
- private int curProgId = config.getInteger("ProgramId", -1);
- private String resultPath;
- private String expectedResult;
+ private String resultPath;
+ private String expectedResult;
- public TestGraphOperations(Configuration config) {
- super(config);
- }
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
- @Override
- protected void preSubmit() throws Exception {
- resultPath = getTempDirPath("result");
+ @Before
+ public void before() throws Exception{
+ resultPath = tempFolder.newFile().toURI().toString();
}
- @Override
- protected void testProgram() throws Exception {
- expectedResult = GraphProgs.runProgram(curProgId, resultPath);
+ @After
+ public void after() throws Exception{
+ compareResultsByLinesInMemory(expectedResult, resultPath);
}
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(expectedResult, resultPath);
+ @Test
+ public void testUndirected() throws Exception {
+ /*
+ * Test getUndirected()
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+
+ graph.getUndirected().getEdges().writeAsCsv(resultPath);
+ env.execute();
+ 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" +
+ "3,4,34\n" + "4,3,34\n" +
+ "3,5,35\n" + "5,3,35\n" +
+ "4,5,45\n" + "5,4,45\n" +
+ "5,1,51\n" + "1,5,51\n";
}
- @Parameters
- public static Collection<Object[]> getConfigurations() throws FileNotFoundException, IOException {
+ @Test
+ public void testReverse() throws Exception {
+ /*
+ * Test reverse()
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+
+ graph.reverse().getEdges().writeAsCsv(resultPath);
+ env.execute();
+ expectedResult = "2,1,12\n" +
+ "3,1,13\n" +
+ "3,2,23\n" +
+ "4,3,34\n" +
+ "5,3,35\n" +
+ "5,4,45\n" +
+ "1,5,51\n";
+ }
- LinkedList<Configuration> tConfigs = new LinkedList<Configuration>();
+ @SuppressWarnings("serial")
+ @Test
+ public void testSubGraph() throws Exception {
+ /*
+ * Test subgraph:
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+ graph.subgraph(new FilterFunction<Vertex<Long, Long>>() {
+ public boolean filter(Vertex<Long, Long> vertex) throws Exception {
+ return (vertex.getValue() > 2);
+ }
+ },
+ new FilterFunction<Edge<Long, Long>>() {
+ public boolean filter(Edge<Long, Long> edge) throws Exception {
+ return (edge.getValue() > 34);
+ }
+ }).getEdges().writeAsCsv(resultPath);
+
+ env.execute();
+ expectedResult = "3,5,35\n" +
+ "4,5,45\n";
+ }
- for(int i=1; i <= NUM_PROGRAMS; i++) {
- Configuration config = new Configuration();
- config.setInteger("ProgramId", i);
- tConfigs.add(config);
- }
+ @SuppressWarnings("serial")
+ @Test
+ public void testFilterVertices() throws Exception {
+ /*
+ * Test filterOnVertices:
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+ graph.filterOnVertices(new FilterFunction<Vertex<Long, Long>>() {
+ public boolean filter(Vertex<Long, Long> vertex) throws Exception {
+ return (vertex.getValue() > 2);
+ }
+ }).getEdges().writeAsCsv(resultPath);
- return toParameterList(tConfigs);
+ env.execute();
+ expectedResult = "3,4,34\n" +
+ "3,5,35\n" +
+ "4,5,45\n";
}
- private static class GraphProgs {
-
- @SuppressWarnings("serial")
- public static String runProgram(int progId, String resultPath) throws Exception {
-
- switch(progId) {
- case 1: {
- /*
- * Test getUndirected()
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- TestGraphUtils.getLongLongEdgeData(env), env);
-
- graph.getUndirected().getEdges().writeAsCsv(resultPath);
- env.execute();
- return "1,2,12\n" + "2,1,12\n" +
- "1,3,13\n" + "3,1,13\n" +
- "2,3,23\n" + "3,2,23\n" +
- "3,4,34\n" + "4,3,34\n" +
- "3,5,35\n" + "5,3,35\n" +
- "4,5,45\n" + "5,4,45\n" +
- "5,1,51\n" + "1,5,51\n";
- }
- case 2: {
- /*
- * Test reverse()
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- TestGraphUtils.getLongLongEdgeData(env), env);
-
- graph.reverse().getEdges().writeAsCsv(resultPath);
- env.execute();
- return "2,1,12\n" +
- "3,1,13\n" +
- "3,2,23\n" +
- "4,3,34\n" +
- "5,3,35\n" +
- "5,4,45\n" +
- "1,5,51\n";
- }
- case 3: {
- /*
- * Test subgraph:
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- TestGraphUtils.getLongLongEdgeData(env), env);
- graph.subgraph(new FilterFunction<Vertex<Long, Long>>() {
- public boolean filter(Vertex<Long, Long> vertex) throws Exception {
- return (vertex.getValue() > 2);
- }
- },
- new FilterFunction<Edge<Long, Long>>() {
- public boolean filter(Edge<Long, Long> edge) throws Exception {
- return (edge.getValue() > 34);
- }
- }).getEdges().writeAsCsv(resultPath);
-
- env.execute();
- return "3,5,35\n" +
- "4,5,45\n";
- }
- case 4: {
- /*
- * Test filterOnVertices:
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- TestGraphUtils.getLongLongEdgeData(env), env);
- graph.filterOnVertices(new FilterFunction<Vertex<Long, Long>>() {
- public boolean filter(Vertex<Long, Long> vertex) throws Exception {
- return (vertex.getValue() > 2);
- }
- }).getEdges().writeAsCsv(resultPath);
-
- env.execute();
- return "3,4,34\n" +
- "3,5,35\n" +
- "4,5,45\n";
- }
- case 5: {
- /*
- * Test filterOnEdges:
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- TestGraphUtils.getLongLongEdgeData(env), env);
- graph.filterOnEdges(new FilterFunction<Edge<Long, Long>>() {
- public boolean filter(Edge<Long, Long> edge) throws Exception {
- return (edge.getValue() > 34);
- }
- }).getEdges().writeAsCsv(resultPath);
-
- env.execute();
- return "3,5,35\n" +
- "4,5,45\n" +
- "5,1,51\n";
- }
- case 6: {
- /*
- * Test numberOfVertices()
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- TestGraphUtils.getLongLongEdgeData(env), env);
- graph.numberOfVertices().writeAsText(resultPath);
-
- env.execute();
- return "5";
- }
- case 7: {
- /*
- * Test numberOfEdges()
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- TestGraphUtils.getLongLongEdgeData(env), env);
- graph.numberOfEdges().writeAsText(resultPath);
-
- env.execute();
- return "7";
- }
- case 8: {
- /*
- * Test getVertexIds()
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- TestGraphUtils.getLongLongEdgeData(env), env);
- graph.getVertexIds().writeAsText(resultPath);
-
- env.execute();
- return "1\n2\n3\n4\n5\n";
- }
- case 9: {
- /*
- * Test getEdgeIds()
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- TestGraphUtils.getLongLongEdgeData(env), env);
- graph.getEdgeIds().writeAsCsv(resultPath);
-
- env.execute();
- return "1,2\n" + "1,3\n" +
- "2,3\n" + "3,4\n" +
- "3,5\n" + "4,5\n" +
- "5,1\n";
- }
- case 10: {
- /*
- * Test union()
- */
-
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
- TestGraphUtils.getLongLongEdgeData(env), env);
-
- List<Vertex<Long, Long>> vertices = new ArrayList<Vertex<Long, Long>>();
- List<Edge<Long, Long>> edges = new ArrayList<Edge<Long, Long>>();
-
- vertices.add(new Vertex<Long, Long>(6L, 6L));
- edges.add(new Edge<Long, Long>(6L, 1L, 61L));
-
- graph = graph.union(Graph.fromCollection(vertices, edges, env));
-
- graph.getEdges().writeAsCsv(resultPath);
-
- env.execute();
-
- return "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" +
- "6,1,61\n";
- }
- default:
- throw new IllegalArgumentException("Invalid program id");
+ @SuppressWarnings("serial")
+ @Test
+ public void testFilterEdges() throws Exception {
+ /*
+ * Test filterOnEdges:
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+ graph.filterOnEdges(new FilterFunction<Edge<Long, Long>>() {
+ public boolean filter(Edge<Long, Long> edge) throws Exception {
+ return (edge.getValue() > 34);
}
- }
+ }).getEdges().writeAsCsv(resultPath);
+
+ env.execute();
+ expectedResult = "3,5,35\n" +
+ "4,5,45\n" +
+ "5,1,51\n";
+ }
+
+ @Test
+ public void testNumberOfVertices() throws Exception {
+ /*
+ * Test numberOfVertices()
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+ graph.numberOfVertices().writeAsText(resultPath);
+
+ env.execute();
+ expectedResult = "5";
+ }
+
+ @Test
+ public void testNumberOfEdges() throws Exception {
+ /*
+ * Test numberOfEdges()
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+ graph.numberOfEdges().writeAsText(resultPath);
+
+ env.execute();
+ expectedResult = "7";
+ }
+
+ @Test
+ public void testVertexIds() throws Exception {
+ /*
+ * Test getVertexIds()
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+ graph.getVertexIds().writeAsText(resultPath);
+
+ env.execute();
+ expectedResult = "1\n2\n3\n4\n5\n";
+ }
+
+ @Test
+ public void testEdgesIds() throws Exception {
+ /*
+ * Test getEdgeIds()
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+ graph.getEdgeIds().writeAsCsv(resultPath);
+
+ env.execute();
+ expectedResult = "1,2\n" + "1,3\n" +
+ "2,3\n" + "3,4\n" +
+ "3,5\n" + "4,5\n" +
+ "5,1\n";
}
-}
+ @Test
+ public void testUnion() throws Exception {
+ /*
+ * Test union()
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+ TestGraphUtils.getLongLongEdgeData(env), env);
+
+ List<Vertex<Long, Long>> vertices = new ArrayList<Vertex<Long, Long>>();
+ List<Edge<Long, Long>> edges = new ArrayList<Edge<Long, Long>>();
+
+ vertices.add(new Vertex<Long, Long>(6L, 6L));
+ edges.add(new Edge<Long, Long>(6L, 1L, 61L));
+
+ graph = graph.union(Graph.fromCollection(vertices, edges, env));
+
+ graph.getEdges().writeAsCsv(resultPath);
+
+ env.execute();
+
+ 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" +
+ "6,1,61\n";
+ }
+}
\ No newline at end of file