aboutsummaryrefslogtreecommitdiff
path: root/graphx
diff options
context:
space:
mode:
authorluluorta <luluorta@gmail.com>2014-11-01 01:22:46 -0700
committerReynold Xin <rxin@databricks.com>2014-11-01 01:22:46 -0700
commitee29ef3800438501e0ff207feb00a28973fc0769 (patch)
treec2eba09ecbb3866fc424b941600ab97225ce3388 /graphx
parentf4e0b28c859412ec8bdfdf452b6a1b2e1bee310e (diff)
downloadspark-ee29ef3800438501e0ff207feb00a28973fc0769.tar.gz
spark-ee29ef3800438501e0ff207feb00a28973fc0769.tar.bz2
spark-ee29ef3800438501e0ff207feb00a28973fc0769.zip
[SPARK-4115][GraphX] Add overrided count for edge counting of EdgeRDD.
Accumulate sizes of all the EdgePartitions just like the VertexRDD. Author: luluorta <luluorta@gmail.com> Closes #2975 from luluorta/graph-edge-count and squashes the following commits: 86ef0e5 [luluorta] Add overrided count for edge counting of EdgeRDD.
Diffstat (limited to 'graphx')
-rw-r--r--graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala7
1 files changed, 6 insertions, 1 deletions
diff --git a/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala b/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
index 5bcb96b136..5267560b3e 100644
--- a/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
+++ b/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
@@ -82,12 +82,17 @@ class EdgeRDD[@specialized ED: ClassTag, VD: ClassTag](
this
}
- /** Persists the vertex partitions using `targetStorageLevel`, which defaults to MEMORY_ONLY. */
+ /** Persists the edge partitions using `targetStorageLevel`, which defaults to MEMORY_ONLY. */
override def cache(): this.type = {
partitionsRDD.persist(targetStorageLevel)
this
}
+ /** The number of edges in the RDD. */
+ override def count(): Long = {
+ partitionsRDD.map(_._2.size.toLong).reduce(_ + _)
+ }
+
private[graphx] def mapEdgePartitions[ED2: ClassTag, VD2: ClassTag](
f: (PartitionID, EdgePartition[ED, VD]) => EdgePartition[ED2, VD2]): EdgeRDD[ED2, VD2] = {
this.withPartitionsRDD[ED2, VD2](partitionsRDD.mapPartitions({ iter =>