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

Reuse previous localVidMap if available

Browse source
This commit is contained in:
Ankur Dave 2013-12-13 02:03:08 -08:00
parent 5e20cbaf66
commit 8f4b8e9b95
1 changed files with 20 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
graph/src/main/scala/org/apache/spark/graph/impl/VTableReplicated.scala
View file

@ -23,21 +23,26 @@ class VTableReplicated[VD: ClassManifest](
vertexPlacement: VertexPlacement, vertexPlacement: VertexPlacement,
prevVTableReplicated: Option[VTableReplicated[VD]] = None) { prevVTableReplicated: Option[VTableReplicated[VD]] = None) {
// Within each edge partition, create a local map from vid to an index into /**
// the attribute array. Each map contains a superset of the vertices that it * Within each edge partition, create a local map from vid to an index into the attribute
// will receive, because it stores vids from both the source and destination * array. Each map contains a superset of the vertices that it will receive, because it stores
// of edges. It must always include both source and destination vids because * vids from both the source and destination of edges. It must always include both source and
// some operations, such as GraphImpl.mapReduceTriplets, rely on this. * destination vids because some operations, such as GraphImpl.mapReduceTriplets, rely on this.
val localVidMap: RDD[(Int, VertexIdToIndexMap)] = edges.partitionsRDD.mapPartitions(_.map { */
case (pid, epart) => private val localVidMap: RDD[(Int, VertexIdToIndexMap)] = prevVTableReplicated match {
val vidToIndex = new VertexIdToIndexMap case Some(prev) =>
epart.foreach { e => prev.localVidMap
vidToIndex.add(e.srcId) case None =>
vidToIndex.add(e.dstId) edges.partitionsRDD.mapPartitions(_.map {
} case (pid, epart) =>
(pid, vidToIndex) val vidToIndex = new VertexIdToIndexMap
}, preservesPartitioning = true).cache() epart.foreach { e =>
vidToIndex.add(e.srcId)
vidToIndex.add(e.dstId)
}
(pid, vidToIndex)
}, preservesPartitioning = true).cache()
}
val bothAttrs: RDD[(Pid, VertexPartition[VD])] = createVTableReplicated(true, true) val bothAttrs: RDD[(Pid, VertexPartition[VD])] = createVTableReplicated(true, true)
val srcAttrOnly: RDD[(Pid, VertexPartition[VD])] = createVTableReplicated(true, false) val srcAttrOnly: RDD[(Pid, VertexPartition[VD])] = createVTableReplicated(true, false)