Create a copy of the Edge objects in EdgeRDD.compute(). This avoids exposing the object re-use, while still enables the more efficient behavior for internal code.

darabos · darabos · commit 49556977bab6 · 2014-04-01T13:05:04.000+02:00
diff --git a/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala b/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
@@ -45,7 +45,8 @@ class EdgeRDD[@specialized ED: ClassTag](
     partitionsRDD.partitioner.orElse(Some(Partitioner.defaultPartitioner(partitionsRDD)))
 
   override def compute(part: Partition, context: TaskContext): Iterator[Edge[ED]] = {
-    firstParent[(PartitionID, EdgePartition[ED])].iterator(part, context).next._2.iterator
+    val p = firstParent[(PartitionID, EdgePartition[ED])].iterator(part, context)
+    p.next._2.iterator.map(_.copy())
   }
 
   override def collect(): Array[Edge[ED]] = this.map(_.copy()).collect()

Original file line number	Diff line number	Diff line change
`@@ -45,7 +45,8 @@ class EdgeRDD[@specialized ED: ClassTag](`
`45`	`45`	`partitionsRDD.partitioner.orElse(Some(Partitioner.defaultPartitioner(partitionsRDD)))`
`46`	`46`
`47`	`47`	`override def compute(part: Partition, context: TaskContext): Iterator[Edge[ED]] = {`
`48`		`- firstParent[(PartitionID, EdgePartition[ED])].iterator(part, context).next._2.iterator`
	`48`	`+ val p = firstParent[(PartitionID, EdgePartition[ED])].iterator(part, context)`
	`49`	`+ p.next._2.iterator.map(_.copy())`
`49`	`50`	`}`
`50`	`51`
`51`	`52`	`override def collect(): Array[Edge[ED]] = this.map(_.copy()).collect()`