Spark: GraphX не может найти связанные компоненты в графах с несколькими ребрами и длинными путями
Я новичок в Spark и GraphX и провел несколько экспериментов с его алгоритмом, чтобы найти связанные компоненты. Я заметил, что структура графика, похоже, сильно влияет на производительность.
Он был в состоянии вычислить графы с миллионами вершин и ребер, но для определенной группы графов алгоритм не завершился вовремя, но в конечном итоге потерпел неудачу с OutOfMemoryError: GC overhead limit exceeded,
Алгоритм, похоже, имеет проблемы с графами, которые содержат длинные пути. Например, для этого графика { (i,i+1) | i <- {1..200} } вычисления не удаются. Однако, когда я добавил транзитивные ребра, вычисление сразу закончилось:
{ (i,j) | i <- {1..200}, j <- {i+1,200} }
Также графики, как это не было проблемой:
{ (i,1) | i <- {1..200} }
Вот минимальный пример для воспроизведения проблемы:
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 =>
arg.dropWhile(_ == '-').split('=') match {
case Array(opt, v) => opt -> v
case _ => 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()
}
}
input.graph Файл может выглядеть вот так (10 узлов, 9 ребер, соединяющих их):
1 2
2 3
3 4
4 5
5 6
6 7
7 8
8 9
9 10
Когда это терпит неудачу, это висит в ConnectedComponents.run(graph), Сообщение об ошибке:
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.<init>(Stage.scala:99)
at org.apache.spark.scheduler.ShuffleMapStage.<init>(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)
Я запускаю локальный узел Spark и запускаю JVM со следующими параметрами:
-Dspark.master=local -Dspark.local.dir=/home/phil/tmp/spark-tmp -Xms8g -Xmx8g
Можете ли вы помочь мне понять, почему у него есть проблема с этим игрушечным графом (201 узел и 200 ребер), но с другой стороны, можно решить реалистичный граф с несколькими миллионами ребер примерно за 80 секунд? (В обоих примерах я использую одинаковые настройки и настройки.)
ОБНОВИТЬ:
Также могут быть воспроизведены в искровой оболочке:
import org.apache.spark.graphx._
import org.apache.spark.graphx.lib._
val graph = GraphLoader.edgeListFile(sc, "input.graph").cache()
ConnectedComponents.run(graph)
Я создал отчет об ошибке: SPARK-15042
1 ответ
Согласно SPARK-15042, проблема все еще существует в 2.1.0-снимке.
Прогресс в исправлении ошибки можно увидеть в SPARK-5484.