Skip to content

Prefer VertexRDD over VertexRDDImpl #56

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
benmccann wants to merge 2 commits into from
Closed

Prefer VertexRDD over VertexRDDImpl #56

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3a14bc8
Prefer VertexRDD to VertexRDDImpl
Jul 20, 2016
71da0a9
Make a few EdgePartition methods private to match upstream
Jul 20, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ml/src/main/scala/com/github/cloudml/zen/ml/clustering/LDA.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ import org.apache.spark.storage.StorageLevel


class LDA(@transient var edges: EdgeRDDImpl[TA, _],
@transient var verts: VertexRDDImpl[TC],
@transient var verts: VertexRDD[TC],
val numTopics: Int,
val numTerms: Int,
val numDocs: Long,
Expand Down
13 changes: 7 additions & 6 deletions ml/src/main/scala/com/github/cloudml/zen/ml/clustering/algorithm/LDAAlgorithm.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import com.github.cloudml.zen.ml.clustering.LDADefines._
import com.github.cloudml.zen.ml.clustering.{LDALogLikelihood, LDAPerplexity}
import com.github.cloudml.zen.ml.util.BVDecompressor
import com.github.cloudml.zen.ml.util.Concurrent._
import org.apache.spark.graphx2.VertexRDD
import org.apache.spark.graphx2.impl.{ShippableVertexPartition => VertPartition, _}

import scala.collection.JavaConversions._
Expand Down Expand Up @@ -74,7 +75,7 @@ abstract class LDAAlgorithm(numTopics: Int,
}

def sampleGraph(edges: EdgeRDDImpl[TA, _],
verts: VertexRDDImpl[TC],
verts: VertexRDD[TC],
topicCounters: BDV[Count],
seed: Int,
sampIter: Int,
Expand All @@ -98,7 +99,7 @@ abstract class LDAAlgorithm(numTopics: Int,
}

def updateVertexCounters(edges: EdgeRDDImpl[TA, Int],
verts: VertexRDDImpl[TC]): VertexRDDImpl[TC] = {
verts: VertexRDD[TC]): VertexRDD[TC] = {
val shippedCounters = edges.partitionsRDD.mapPartitions(_.flatMap { case (_, ep) =>
countPartition(ep)
}).partitionBy(verts.partitioner.get)
Expand All @@ -112,7 +113,7 @@ abstract class LDAAlgorithm(numTopics: Int,
}

def calcPerplexity(edges: EdgeRDDImpl[TA, _],
verts: VertexRDDImpl[TC],
verts: VertexRDD[TC],
topicCounters: BDV[Count],
numTokens: Long,
numTerms: Int,
Expand All @@ -131,7 +132,7 @@ abstract class LDAAlgorithm(numTopics: Int,
new LDAPerplexity(pplx, wpplx, dpplx)
}

def calcLogLikelihood(verts: VertexRDDImpl[TC],
def calcLogLikelihood(verts: VertexRDD[TC],
topicCounters: BDV[Count],
numTokens: Long,
numDocs: Long,
Expand All @@ -151,7 +152,7 @@ abstract class LDAAlgorithm(numTopics: Int,
}

def refreshEdgeAssociations(edges: EdgeRDDImpl[TA, _],
verts: VertexRDDImpl[TC]): EdgeRDDImpl[TA, Nvk] = {
verts: VertexRDD[TC]): EdgeRDDImpl[TA, Nvk] = {
val shippedVerts = verts.partitionsRDD.mapPartitions(_.flatMap { vp =>
val rt = vp.routingTable
val index = vp.index
Expand Down Expand Up @@ -201,7 +202,7 @@ abstract class LDAAlgorithm(numTopics: Int,
edges.withPartitionsRDD(partRDD)
}

def collectTopicCounters(verts: VertexRDDImpl[TC]): BDV[Count] = {
def collectTopicCounters(verts: VertexRDD[TC]): BDV[Count] = {
verts.partitionsRDD.mapPartitions(_.map { vp =>
val totalSize = vp.capacity
val index = vp.index
Expand Down
6 changes: 3 additions & 3 deletions ml/src/main/scala/org/apache/spark/graphx2/impl/EdgePartition.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,11 +109,11 @@ class EdgePartition[
activeSet)
}

@inline def srcIds(pos: Int): VertexId = local2global(localSrcIds(pos))
@inline private def srcIds(pos: Int): VertexId = local2global(localSrcIds(pos))

@inline def dstIds(pos: Int): VertexId = local2global(localDstIds(pos))
@inline private def dstIds(pos: Int): VertexId = local2global(localDstIds(pos))

@inline def attrs(pos: Int): ED = data(pos)
@inline private def attrs(pos: Int): ED = data(pos)

/** Look up vid in activeSet, throwing an exception if it is None. */
def isActive(vid: VertexId): Boolean = {
Expand Down
2 changes: 1 addition & 1 deletion ml/src/main/scala/org/apache/spark/graphx2/impl/VertexRDDImpl.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -233,7 +233,7 @@ class VertexRDDImpl[VD] (
}

override def withPartitionsRDD[VD2: ClassTag](
partitionsRDD: RDD[ShippableVertexPartition[VD2]]): VertexRDDImpl[VD2] = {
partitionsRDD: RDD[ShippableVertexPartition[VD2]]): VertexRDD[VD2] = {
new VertexRDDImpl(partitionsRDD, this.targetStorageLevel)
}

Expand Down