Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/3054#discussion_r20124915
--- Diff:
graphx/src/main/scala/org/apache/spark/graphx/impl/EdgePartitionBuilder.scala
---
@@ -38,19 +39,77 @@ class EdgePartitionBuilder[@specialized(Long, Int,
Double) ED: ClassTag, VD: Cla
def toEdgePartition: EdgePartition[ED, VD] = {
val edgeArray = edges.trim().array
Sorting.quickSort(edgeArray)(Edge.lexicographicOrdering)
- val srcIds = new Array[VertexId](edgeArray.size)
- val dstIds = new Array[VertexId](edgeArray.size)
+ val localSrcIds = new Array[Int](edgeArray.size)
+ val localDstIds = new Array[Int](edgeArray.size)
+ val data = new Array[ED](edgeArray.size)
+ val index = new GraphXPrimitiveKeyOpenHashMap[VertexId, Int]
+ val global2local = new GraphXPrimitiveKeyOpenHashMap[VertexId, Int]
+ val local2global = new PrimitiveVector[VertexId]
+ var vertexAttrs = Array.empty[VD]
+ // Copy edges into columnar structures, tracking the beginnings of
source vertex id clusters and
+ // adding them to the index. Also populate a map from vertex id to a
sequential local offset.
+ if (edgeArray.length > 0) {
+ index.update(edgeArray(0).srcId, 0)
+ var currSrcId: VertexId = edgeArray(0).srcId
+ var currLocalId = -1
+ var i = 0
+ while (i < edgeArray.size) {
+ val srcId = edgeArray(i).srcId
+ val dstId = edgeArray(i).dstId
+ localSrcIds(i) = global2local.changeValue(srcId,
+ { currLocalId += 1; local2global += srcId; currLocalId },
identity)
+ localDstIds(i) = global2local.changeValue(dstId,
+ { currLocalId += 1; local2global += dstId; currLocalId },
identity)
+ data(i) = edgeArray(i).attr
+ if (srcId != currSrcId) {
+ currSrcId = srcId
+ index.update(currSrcId, i)
+ }
+
+ i += 1
+ }
+ vertexAttrs = new Array[VD](currLocalId + 1)
+ }
+ new EdgePartition(
+ localSrcIds, localDstIds, data, index, global2local,
local2global.trim().array, vertexAttrs)
+ }
+}
+
+/**
+ * Constructs an EdgePartition from an existing EdgePartition with the
same vertex set. This enables
+ * reuse of the local vertex ids. Intended for internal use in
EdgePartition only.
+ */
+private[impl]
+class ExistingEdgePartitionBuilder[
+ @specialized(Long, Int, Double) ED: ClassTag, VD: ClassTag](
+ global2local: GraphXPrimitiveKeyOpenHashMap[VertexId, Int],
+ local2global: Array[VertexId],
+ vertexAttrs: Array[VD],
+ activeSet: Option[VertexSet],
+ size: Int = 64) {
+ var edges = new PrimitiveVector[EdgeWithLocalIds[ED]](size)
--- End diff --
also do it for the one in EdgePartitionBuilder
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
