diff options
author | Reynold Xin <rxin@apache.org> | 2014-01-01 13:28:34 -0800 |
---|---|---|
committer | Reynold Xin <rxin@apache.org> | 2014-01-01 13:28:34 -0800 |
commit | dc9cb83e8661d075d8f3b47c27b2f423287e598d (patch) | |
tree | b444f923e0b5e21470e4a3a8bf7c10f927b79aae /graph/src | |
parent | 32d6ae9d9c15bdb7937e7bd69a55afb21631adb6 (diff) | |
parent | 2f2524fd11876a19e315f994020877564a8a0df8 (diff) | |
download | spark-dc9cb83e8661d075d8f3b47c27b2f423287e598d.tar.gz spark-dc9cb83e8661d075d8f3b47c27b2f423287e598d.tar.bz2 spark-dc9cb83e8661d075d8f3b47c27b2f423287e598d.zip |
Merge pull request #126 from jegonzal/FixingPersist
Fixing Persist Behavior
Diffstat (limited to 'graph/src')
-rw-r--r-- | graph/src/main/scala/org/apache/spark/graph/EdgeRDD.scala | 6 | ||||
-rw-r--r-- | graph/src/main/scala/org/apache/spark/graph/VertexRDD.scala | 2 |
2 files changed, 4 insertions, 4 deletions
diff --git a/graph/src/main/scala/org/apache/spark/graph/EdgeRDD.scala b/graph/src/main/scala/org/apache/spark/graph/EdgeRDD.scala index 63858db2ef..9aa76c9394 100644 --- a/graph/src/main/scala/org/apache/spark/graph/EdgeRDD.scala +++ b/graph/src/main/scala/org/apache/spark/graph/EdgeRDD.scala @@ -22,9 +22,8 @@ class EdgeRDD[@specialized ED: ClassManifest]( override val partitioner = partitionsRDD.partitioner.orElse(Some(Partitioner.defaultPartitioner(partitionsRDD))) - override def compute(split: Partition, context: TaskContext): Iterator[Edge[ED]] = { - val edgePartition = partitionsRDD.compute(split, context).next()._2 - edgePartition.iterator + override def compute(part: Partition, context: TaskContext): Iterator[Edge[ED]] = { + firstParent[(Pid, EdgePartition[ED])].iterator(part, context).next._2.iterator } override def collect(): Array[Edge[ED]] = this.map(_.copy()).collect() @@ -45,6 +44,7 @@ class EdgeRDD[@specialized ED: ClassManifest]( def mapEdgePartitions[ED2: ClassManifest](f: EdgePartition[ED] => EdgePartition[ED2]) : EdgeRDD[ED2] = { +// iter => iter.map { case (pid, ep) => (pid, f(ep)) } new EdgeRDD[ED2](partitionsRDD.mapPartitions({ iter => val (pid, ep) = iter.next() Iterator(Tuple2(pid, f(ep))) diff --git a/graph/src/main/scala/org/apache/spark/graph/VertexRDD.scala b/graph/src/main/scala/org/apache/spark/graph/VertexRDD.scala index fe0f0ae491..c274e342c7 100644 --- a/graph/src/main/scala/org/apache/spark/graph/VertexRDD.scala +++ b/graph/src/main/scala/org/apache/spark/graph/VertexRDD.scala @@ -105,7 +105,7 @@ class VertexRDD[@specialized VD: ClassManifest]( * Provide the `RDD[(Vid, VD)]` equivalent output. */ override def compute(part: Partition, context: TaskContext): Iterator[(Vid, VD)] = { - partitionsRDD.compute(part, context).next().iterator + firstParent[VertexPartition[VD]].iterator(part, context).next.iterator } /** |