Skip to content

Unify GraphImpl RDDs + other graph load optimizations #497

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
ankurdave wants to merge 17 commits into from
Closed

Unify GraphImpl RDDs + other graph load optimizations #497

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
d64e8d4
Log current Pregel iteration
ankurdave Apr 20, 2014
62c7b78
In Analytics, take PageRank numIter
ankurdave Apr 20, 2014
d6d60e2
In GraphLoader, coalesce to minEdgePartitions
ankurdave Apr 20, 2014
e4fbd32
Unify GraphImpl RDDs + other graph load optimizations
ankurdave Apr 13, 2014
ae36110
Fix style errors
ankurdave Apr 23, 2014
10b3596
Clean up public API
ankurdave Apr 23, 2014
2a928b2
Set locality wait
ankurdave Apr 25, 2014
0d3584c
EdgePartition.size should be val
ankurdave Apr 25, 2014
c88b269
Revert upgradeIterator to if-in-a-loop
ankurdave May 7, 2014
04d3ae5
Convert implicit parameter to context bound
ankurdave May 7, 2014
75af062
Add explicit return types
ankurdave May 7, 2014
57202e8
Replace case with pair parameter
ankurdave May 7, 2014
a04765c
Remove unnecessary toOps call
ankurdave May 7, 2014
13ac845
Merge remote-tracking branch 'apache-spark/master' into unify-rdds
ankurdave May 7, 2014
5ba8789
Add GraphX upgrade guide from Spark 0.9.1
ankurdave May 7, 2014
4933e2e
Exclude RoutingTable from binary compatibility check
ankurdave May 7, 2014
332ab43
Merge remote-tracking branch 'apache-spark/master' into unify-rdds
ankurdave May 10, 2014
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
Prev Previous commit
Next Next commit
Fix style errors
  • Loading branch information
@ankurdave
ankurdave committed Apr 23, 2014
commit ae361108c2885d173cdc73b3124ad7b40e9fdfdf
3 changes: 2 additions & 1 deletion graphx/src/main/scala/org/apache/spark/graphx/VertexRDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -396,7 +396,8 @@ object VertexRDD {
* @param numPartitions the desired number of partitions for the resulting `VertexRDD`
* @param defaultVal the vertex attribute to use when creating missing vertices
*/
def fromEdges[VD: ClassTag](edges: EdgeRDD[_, _], numPartitions: Int, defaultVal: VD): VertexRDD[VD] = {
def fromEdges[VD: ClassTag](
edges: EdgeRDD[_, _], numPartitions: Int, defaultVal: VD): VertexRDD[VD] = {
val routingTables = createRoutingTables(edges, new HashPartitioner(numPartitions))
val vertexPartitions = routingTables.mapPartitions({ routingTableIter =>
val routingTable =
Expand Down
3 changes: 2 additions & 1 deletion graphx/src/main/scala/org/apache/spark/graphx/impl/EdgePartition.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,8 @@ import org.apache.spark.graphx.util.collection.PrimitiveKeyOpenHashMap
* @param activeSet an optional active vertex set for filtering computation on the edges
*/
private[graphx]
class EdgePartition[@specialized(Char, Int, Boolean, Byte, Long, Float, Double) ED: ClassTag, VD: ClassTag](
class EdgePartition[
@specialized(Char, Int, Boolean, Byte, Long, Float, Double) ED: ClassTag, VD: ClassTag](
Copy link
Contributor Author

@ankurdave ankurdave May 8, 2014

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

specialize VD

val srcIds: Array[VertexId],
val dstIds: Array[VertexId],
val data: Array[ED],
Expand Down
4 changes: 2 additions & 2 deletions graphx/src/main/scala/org/apache/spark/graphx/impl/GraphImpl.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -230,8 +230,8 @@ class GraphImpl[VD: ClassTag, ED: ClassTag] protected (
// updateF preserves type, so we can use incremental replication
val newVerts = vertices.leftJoin(other)(updateF).cache()
val changedVerts = vertices.asInstanceOf[VertexRDD[VD2]].diff(newVerts)
val newReplicatedVertexView =
replicatedVertexView.asInstanceOf[ReplicatedVertexView[VD2, ED]].updateVertices(changedVerts)
val newReplicatedVertexView = replicatedVertexView.asInstanceOf[ReplicatedVertexView[VD2, ED]]
.updateVertices(changedVerts)
new GraphImpl(newVerts, newReplicatedVertexView)
} else {
// updateF does not preserve type, so we must re-replicate all vertices
Expand Down
3 changes: 2 additions & 1 deletion graphx/src/main/scala/org/apache/spark/graphx/impl/ReplicatedVertexView.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,8 @@ class ReplicatedVertexView[VD: ClassTag, ED: ClassTag](
var hasDstId: Boolean = false) {

/**
* Return a new `ReplicatedVertexView` with the specified `EdgeRDD`, which must have the same shipping level.
* Return a new `ReplicatedVertexView` with the specified `EdgeRDD`, which must have the same
* shipping level.
*/
def withEdges[VD2: ClassTag, ED2: ClassTag](
edges_ : EdgeRDD[ED2, VD2]): ReplicatedVertexView[VD2, ED2] = {
Expand Down
4 changes: 2 additions & 2 deletions graphx/src/main/scala/org/apache/spark/graphx/impl/ShippableVertexPartition.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,8 +39,8 @@ object ShippableVertexPartition {
apply(iter, RoutingTablePartition.empty, null.asInstanceOf[VD])

/**
* Construct a `ShippableVertexPartition` from the given vertices with the specified routing table,
* filling in missing vertices mentioned in the routing table using `defaultVal`.
* Construct a `ShippableVertexPartition` from the given vertices with the specified routing
* table, filling in missing vertices mentioned in the routing table using `defaultVal`.
*/
def apply[VD: ClassTag](
iter: Iterator[(VertexId, VD)], routingTable: RoutingTablePartition, defaultVal: VD)
Expand Down