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:43 UTC
[5/5] flink git commit: [FLINK-2093] [gelly] Added difference Method
[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()