title: Spark: GraphX fails to find connected components in graphs with few edges and long paths
body: <p>I'm new to Spark and GraphX and did some experiments with its algorithm to find connected components. I noticed that the structure of the graph seems to have a strong impact on the performance.</p>
  
  <p>It was able to compute graphs with millions of vertices and edges, but for a certain group of graphs, the algorithm did not finish in time, but eventually fails with an <code>OutOfMemoryError: GC overhead limit exceeded</code>.</p>
  
  <p>The algorithm seems to have problems with graphs that contain long paths. For instance, for this graph <code>{ (i,i+1) | i &lt;- {1..200} }</code> the computation fails. However, when I added transitive edges, the computation finished immediately:</p>
  
  <pre><code>{ (i,j) | i &lt;- {1..200}, j &lt;- {i+1,200} }
  </code></pre>
  
  <p>Also graphs like this were no problem:</p>
  
  <pre><code>{ (i,1) | i &lt;- {1..200} }
  </code></pre>
  
  <p>Here is a minimal example to reproduce the problem:</p>
  
  <pre><code>import org.apache.spark._
  import org.apache.spark.graphx._
  import org.apache.spark.graphx.lib._
  import org.apache.spark.storage.StorageLevel
  
  import scala.collection.mutable
  
  object Matching extends Logging {
  
    def main(args: Array[String]): Unit = {
      val fname = "input.graph"
      val optionsList = args.drop(1).map { arg =&gt;
        arg.dropWhile(_ == '-').split('=') match {
          case Array(opt, v) =&gt; opt -&gt; v
          case _ =&gt; throw new IllegalArgumentException("Invalid argument: " + arg)
        }
      }
      val options = mutable.Map(optionsList: _*)
  
      val conf = new SparkConf()
      GraphXUtils.registerKryoClasses(conf)
  
      val partitionStrategy: Option[PartitionStrategy] = options.remove("partStrategy")
        .map(PartitionStrategy.fromString(_))
      val edgeStorageLevel = options.remove("edgeStorageLevel")
        .map(StorageLevel.fromString(_)).getOrElse(StorageLevel.MEMORY_ONLY)
      val vertexStorageLevel = options.remove("vertexStorageLevel")
        .map(StorageLevel.fromString(_)).getOrElse(StorageLevel.MEMORY_ONLY)
  
      val sc = new SparkContext(conf.setAppName("ConnectedComponents(" + fname + ")"))
      val unpartitionedGraph = GraphLoader.edgeListFile(sc, fname,
        edgeStorageLevel = edgeStorageLevel,
        vertexStorageLevel = vertexStorageLevel).cache()
      log.info("Loading graph...")
      val graph = partitionStrategy.foldLeft(unpartitionedGraph)(_.partitionBy(_))
      log.info("Loading graph...done")
  
      log.info("Computing connected components...")
      val cc = ConnectedComponents.run(graph)
      log.info("Computed connected components...done")
  
      sc.stop()
    }
  }
  </code></pre>
  
  <p>The <code>input.graph</code> file can look this this (10 nodes, 9 edges connecting them):</p>
  
  <pre><code>1 2
  2 3
  3 4
  4 5
  5 6
  6 7
  7 8
  8 9
  9 10
  </code></pre>
  
  <p>When it fails, it hangs in <code>ConnectedComponents.run(graph)</code>. The error message is:</p>
  
  <pre><code>Exception in thread "dag-scheduler-event-loop" java.lang.OutOfMemoryError: GC overhead limit exceeded
      at java.util.regex.Pattern.compile(Pattern.java:1054)
      at java.lang.String.replace(String.java:2239)
      at org.apache.spark.util.Utils$.getFormattedClassName(Utils.scala:1632)
      at org.apache.spark.storage.RDDInfo$$anonfun$1.apply(RDDInfo.scala:58)
      at org.apache.spark.storage.RDDInfo$$anonfun$1.apply(RDDInfo.scala:58)
      at scala.Option.getOrElse(Option.scala:121)
      at org.apache.spark.storage.RDDInfo$.fromRdd(RDDInfo.scala:58)
      at org.apache.spark.scheduler.StageInfo$$anonfun$1.apply(StageInfo.scala:80)
      at org.apache.spark.scheduler.StageInfo$$anonfun$1.apply(StageInfo.scala:80)
      at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
      at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
      at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
      at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
      at scala.collection.TraversableLike$class.map(TraversableLike.scala:245)
      at scala.collection.AbstractTraversable.map(Traversable.scala:104)
      at org.apache.spark.scheduler.StageInfo$.fromStage(StageInfo.scala:80)
      at org.apache.spark.scheduler.Stage.&lt;init&gt;(Stage.scala:99)
      at org.apache.spark.scheduler.ShuffleMapStage.&lt;init&gt;(ShuffleMapStage.scala:44)
      at org.apache.spark.scheduler.DAGScheduler.newShuffleMapStage(DAGScheduler.scala:317)
      at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$newOrUsedShuffleStage(DAGScheduler.scala:352)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$getShuffleMapStage$1.apply(DAGScheduler.scala:286)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$getShuffleMapStage$1.apply(DAGScheduler.scala:285)
      at scala.collection.Iterator$class.foreach(Iterator.scala:742)
      at scala.collection.AbstractIterator.foreach(Iterator.scala:1194)
      at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
      at scala.collection.mutable.Stack.foreach(Stack.scala:170)
      at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$getShuffleMapStage(DAGScheduler.scala:285)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$visit$1$1.apply(DAGScheduler.scala:389)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$visit$1$1.apply(DAGScheduler.scala:386)
      at scala.collection.immutable.List.foreach(List.scala:381)
      at org.apache.spark.scheduler.DAGScheduler.visit$1(DAGScheduler.scala:386)
      at org.apache.spark.scheduler.DAGScheduler.getParentStages(DAGScheduler.scala:398)
  </code></pre>
  
  <p>I am running a local Spark node and start the JVM with the following options:</p>
  
  <pre><code>-Dspark.master=local -Dspark.local.dir=/home/phil/tmp/spark-tmp -Xms8g -Xmx8g
  </code></pre>
  
  <p>Can you help me understand why it has problem with this toy graph (201 nodes and 200 edges), but on the other hand can solve a realistic graph with multiple millions of edges in about 80 seconds? (In both examples, I use the same setup and configuration.)</p>
  
  <p><strong>UPDATE:</strong></p>
  
  <p>Can also be reproduced in the spark-shell:</p>
  
  <pre><code>import org.apache.spark.graphx._
  import org.apache.spark.graphx.lib._
  
  val graph = GraphLoader.edgeListFile(sc, "input.graph").cache()
  ConnectedComponents.run(graph)
  </code></pre>
  
  <p>I created a bug report: <a href="https://issues.apache.org/jira/browse/SPARK-15042" rel="nofollow">SPARK-15042</a></p>