You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by an...@apache.org on 2014/10/12 23:18:59 UTC
git commit: SPARK-3716 [GraphX] Update Analytics.scala for
partitionStrategy assignment
Repository: spark
Updated Branches:
refs/heads/branch-1.1 18ef22ab7 -> 5a21e3e7e
SPARK-3716 [GraphX] Update Analytics.scala for partitionStrategy assignment
Previously, when the val partitionStrategy was created it called a function in the Analytics object which was a copy of the PartitionStrategy.fromString() method. This function has been removed, and the assignment of partitionStrategy now uses the PartitionStrategy.fromString method instead. In this way, it better matches the declarations of edge/vertex StorageLevel variables.
Author: NamelessAnalyst <Na...@users.noreply.github.com>
Closes #2569 from NamelessAnalyst/branch-1.1 and squashes the following commits:
c24ff51 [NamelessAnalyst] Update Analytics.scala
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/5a21e3e7
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/5a21e3e7
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/5a21e3e7
Branch: refs/heads/branch-1.1
Commit: 5a21e3e7e97f135c81c664098a723434b910f09d
Parents: 18ef22a
Author: NamelessAnalyst <Na...@users.noreply.github.com>
Authored: Sun Oct 12 14:18:55 2014 -0700
Committer: Ankur Dave <an...@gmail.com>
Committed: Sun Oct 12 14:18:55 2014 -0700
----------------------------------------------------------------------
.../org/apache/spark/graphx/lib/Analytics.scala | 19 ++++---------------
1 file changed, 4 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/5a21e3e7/graphx/src/main/scala/org/apache/spark/graphx/lib/Analytics.scala
----------------------------------------------------------------------
diff --git a/graphx/src/main/scala/org/apache/spark/graphx/lib/Analytics.scala b/graphx/src/main/scala/org/apache/spark/graphx/lib/Analytics.scala
index c1513a0..2752957 100644
--- a/graphx/src/main/scala/org/apache/spark/graphx/lib/Analytics.scala
+++ b/graphx/src/main/scala/org/apache/spark/graphx/lib/Analytics.scala
@@ -45,17 +45,6 @@ object Analytics extends Logging {
}
val options = mutable.Map(optionsList: _*)
- def pickPartitioner(v: String): PartitionStrategy = {
- // TODO: Use reflection rather than listing all the partitioning strategies here.
- v match {
- case "RandomVertexCut" => RandomVertexCut
- case "EdgePartition1D" => EdgePartition1D
- case "EdgePartition2D" => EdgePartition2D
- case "CanonicalRandomVertexCut" => CanonicalRandomVertexCut
- case _ => throw new IllegalArgumentException("Invalid PartitionStrategy: " + v)
- }
- }
-
val conf = new SparkConf()
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
.set("spark.kryo.registrator", "org.apache.spark.graphx.GraphKryoRegistrator")
@@ -66,7 +55,7 @@ object Analytics extends Logging {
sys.exit(1)
}
val partitionStrategy: Option[PartitionStrategy] = options.remove("partStrategy")
- .map(pickPartitioner(_))
+ .map(PartitionStrategy.fromString(_))
val edgeStorageLevel = options.remove("edgeStorageLevel")
.map(StorageLevel.fromString(_)).getOrElse(StorageLevel.MEMORY_ONLY)
val vertexStorageLevel = options.remove("vertexStorageLevel")
@@ -106,7 +95,7 @@ object Analytics extends Logging {
if (!outFname.isEmpty) {
logWarning("Saving pageranks of pages to " + outFname)
- pr.map{case (id, r) => id + "\t" + r}.saveAsTextFile(outFname)
+ pr.map { case (id, r) => id + "\t" + r }.saveAsTextFile(outFname)
}
sc.stop()
@@ -128,7 +117,7 @@ object Analytics extends Logging {
val graph = partitionStrategy.foldLeft(unpartitionedGraph)(_.partitionBy(_))
val cc = ConnectedComponents.run(graph)
- println("Components: " + cc.vertices.map{ case (vid,data) => data}.distinct())
+ println("Components: " + cc.vertices.map { case (vid, data) => data }.distinct())
sc.stop()
case "triangles" =>
@@ -146,7 +135,7 @@ object Analytics extends Logging {
minEdgePartitions = numEPart,
edgeStorageLevel = edgeStorageLevel,
vertexStorageLevel = vertexStorageLevel)
- // TriangleCount requires the graph to be partitioned
+ // TriangleCount requires the graph to be partitioned
.partitionBy(partitionStrategy.getOrElse(RandomVertexCut)).cache()
val triangles = TriangleCount.run(graph)
println("Triangles: " + triangles.vertices.map {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org