ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Cache msgsByPartition

Browse source
This commit is contained in:
Ankur Dave 2013-10-16 23:56:15 -07:00
parent bc234bf0e1
commit 2282d27cf1
1 changed files with 2 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
graph/src/main/scala/org/apache/spark/graph/impl/GraphImpl.scala
View file

@ -486,7 +486,8 @@ object GraphImpl {
.flatMap { case (vid, (vdata, pids)) => .flatMap { case (vid, (vdata, pids)) =>
pids.iterator.map { pid => MessageToPartition(pid, (vid, vdata)) } pids.iterator.map { pid => MessageToPartition(pid, (vid, vdata)) }
} }
.partitionBy(eTable.partitioner.get) //@todo assert edge table has partitioner .partitionBy(eTable.partitioner.get).cache()
// @todo assert edge table has partitioner
val vTableReplicationMap: IndexedRDD[Pid, VertexIdToIndexMap] = val vTableReplicationMap: IndexedRDD[Pid, VertexIdToIndexMap] =
msgsByPartition.mapPartitionsWithIndex( (pid, iter) => { msgsByPartition.mapPartitionsWithIndex( (pid, iter) => {