diff options
author | Ankur Dave <ankurdave@gmail.com> | 2013-12-05 16:15:56 -0800 |
---|---|---|
committer | Ankur Dave <ankurdave@gmail.com> | 2013-12-05 16:15:56 -0800 |
commit | 4d4c1c931e10dd8a5dfcf4c39d6b37d832b382d8 (patch) | |
tree | aa35cd5429c1a4b8ce2bf3f574a06cff0bff111e /graph/src/test | |
parent | 15374fed97588efe23596dd6c8fc96c218bfa140 (diff) | |
download | spark-4d4c1c931e10dd8a5dfcf4c39d6b37d832b382d8.tar.gz spark-4d4c1c931e10dd8a5dfcf4c39d6b37d832b382d8.tar.bz2 spark-4d4c1c931e10dd8a5dfcf4c39d6b37d832b382d8.zip |
Test Graph.deltaJoinVertices and VertexRDD.diff
Diffstat (limited to 'graph/src/test')
-rw-r--r-- | graph/src/test/scala/org/apache/spark/graph/GraphSuite.scala | 30 |
1 files changed, 25 insertions, 5 deletions
diff --git a/graph/src/test/scala/org/apache/spark/graph/GraphSuite.scala b/graph/src/test/scala/org/apache/spark/graph/GraphSuite.scala index 1d23911ff9..d5aa36c04f 100644 --- a/graph/src/test/scala/org/apache/spark/graph/GraphSuite.scala +++ b/graph/src/test/scala/org/apache/spark/graph/GraphSuite.scala @@ -120,9 +120,10 @@ class GraphSuite extends FunSuite with LocalSparkContext { test("VertexSetRDD") { withSpark(new SparkContext("local", "test")) { sc => - val a = sc.parallelize((0 to 100).map(x => (x.toLong, x.toLong)), 5) + val n = 100 + val a = sc.parallelize((0 to n).map(x => (x.toLong, x.toLong)), 5) val b = VertexRDD(a).mapValues(x => -x).cache() // Allow joining b with a derived RDD of b - assert(b.count === 101) + assert(b.count === n + 1) assert(b.leftJoin(a){ (id, a, bOpt) => a + bOpt.get }.map(x=> x._2).reduce(_+_) === 0) val c = b.aggregateUsingIndex[Long](a, (x, y) => x) assert(b.leftJoin(c){ (id, b, cOpt) => b + cOpt.get }.map(x=> x._2).reduce(_+_) === 0) @@ -130,7 +131,8 @@ class GraphSuite extends FunSuite with LocalSparkContext { val e = a.filter(q => ((q._2 % 2) == 0)) assert(d.count === e.count) assert(b.zipJoin(c)((id, b, c) => b + c).map(x => x._2).reduce(_+_) === 0) - + val f = b.mapValues(x => if (x % 2 == 0) -x else x) + assert(b.diff(f).collect().toSet === (2 to n by 2).map(x => (x.toLong, x.toLong)).toSet) } } @@ -143,10 +145,28 @@ class GraphSuite extends FunSuite with LocalSparkContext { val subgraph = star.subgraph(vpred = (vid, attr) => vid % 2 == 0) // We should have 5 vertices. - assert(subgraph.vertices.collect().toSet === (0 to n / 2).map(x => (x * 2, "v")).toSet) + assert(subgraph.vertices.collect().toSet === (0 to n by 2).map(x => (x, "v")).toSet) // And 4 edges. - assert(subgraph.edges.map(_.copy()).collect().toSet === (1 to n / 2).map(x => Edge(0, x * 2, 1)).toSet) + assert(subgraph.edges.map(_.copy()).collect().toSet === (2 to n by 2).map(x => Edge(0, x, 1)).toSet) + } + } + + test("deltaJoinVertices") { + withSpark(new SparkContext("local", "test")) { sc => + // Create a star graph of 10 vertices + val n = 10 + val star = Graph.fromEdgeTuples(sc.parallelize((1 to n).map(x => (0: Vid, x: Vid))), "v1") + + // Modify only vertices whose vids are even + val newVerts = star.vertices.mapValues((vid, attr) => if (vid % 2 == 0) "v2" else attr) + val changedVerts = star.vertices.diff(newVerts) + + // Apply the modification to the graph + val changedStar = star.deltaJoinVertices(newVerts, changedVerts) + + // The graph's vertices should be correct + assert(changedStar.vertices.collect().toSet === newVerts.collect().toSet) } } |