Added a correctEdges function.

graph/src/main/scala/spark/graph/Graph.scala

@@ -141,6 +141,7 @@ abstract class Graph[VD: ClassManifest, ED: ClassManifest] {
   def mapTriplets[ED2: ClassManifest](
     map: EdgeTriplet[VD, ED] => ED2): Graph[VD, ED2]
 
+  def correctEdges(): Graph[VD, ED]
 
   /**
    * Construct a new graph with all the edges reversed.  If this graph contains

graph/src/main/scala/spark/graph/impl/GraphImpl.scala

@@ -91,6 +91,12 @@ class GraphImpl[VD: ClassManifest, ED: ClassManifest] protected (
     newGraph(vertices, triplets.map(e => Edge(e.src.id, e.dst.id, f(e))))
   }
 
+  override def correctEdges(): Graph[VD, ED] = {
+    val sc = vertices.context
+    val vset = sc.broadcast(vertices.map(_.id).collect().toSet)
+    val newEdges = edges.filter(e => vset.value.contains(e.src) && vset.value.contains(e.dst))
+    Graph(vertices, newEdges)
+  }
 
   //////////////////////////////////////////////////////////////////////////////////////////////////
   // Lower level transformation methods