Agfdhyk

I have a problem with the Spark GraphX module. I have a 5 node cluster, with 23.5G Memory and 24 cores per node. I use the spark-shell for submitting my code, so I use Spark in client mode.
In my configuration I have 1 master node and 4 slave nodes. This is my spark-defaults.conf:

spark.executor.instances                8

spark.executor.memory                   10g

spark.driver.memory                     18g

spark.executor.cores                    10

spark.driver.cores                      18

spark.default.parallelism               144

spark.serializer                        org.apache.spark.serializer.KryoSerializer

I read and store 2 really small files with together 40mb file size.

This is my code:

val input1 = sc.textFile("/home/data/spark/nodes.txt")

val vertexArray = input1.map(line => (line.toLong, mutable.Set[VertexId]()))



val input2 = sc.textFile("/home/data/spark/edges.txt")

val splitRdd = input2.map( line => line.split(" ") )

val edgeArray = splitRdd.map(line => Edge(line(0).toLong, line(1).toLong, "bla"))



val vertices: RDD[(VertexId, mutable.Set[VertexId])] = vertexArray

val edges: RDD[Edge[String]] = edgeArray

val graph = Graph(vertices, edges).partitionBy(EdgePartition2D, 4)



var filteredGraph: Graph[mutable.Set[VertexId], String] = graph.mapVertices((vid, vdata) => {

  mutable.Set[VertexId]()

}).cache()

val temp: VertexRDD[mutable.Set[VertexId]] = filteredGraph.aggregateMessages[mutable.Set[VertexId]](

  triplet => {

    if(triplet.dstId < 0){

      triplet.sendToDst(mutable.Set[VertexId](triplet.srcId))

    }

  },

  (oldSet, newSet) => oldSet ++ newSet // Merge Message

)

filteredGraph = filteredGraph.joinVertices(temp)((id, oldSet, newSet) => newSet).cache()

val temp2: VertexRDD[mutable.Set[VertexId]] = filteredGraph.aggregateMessages[mutable.Set[VertexId]](

  triplet => {

    if(triplet.dstId > 0){

      triplet.sendToDst(triplet.srcAttr.filter(id=>triplet.dstId!=id && triplet.dstId < id))

    }

  },

  (oldSet, newSet) => oldSet ++ newSet // Merge Message

)

val candidatesRDD: RDD[(Long, List[Long])] = temp2.map(vertex => {

  (vertex._1.asInstanceOf[Long], vertex._2.asInstanceOf[ mutable.Set[Long]].toList)

})





val newNames = Seq("rid", "candidates")

val candidatesDF = candidatesRDD.toDF(newNames: _*)

val candidatesDFMod = candidatesDF.withColumn("candidates", explode($"candidates"))

candidatesDFMod.show

If I do the computation, I get after several time a java.lang.OutOfMemoryError: Java heap space Exeception of one executor. After this Spark tries to compute it again, so it restarts the stages, but it will end up in the same exception again. Why does this happen? The computation fills the complete 10G executor memory. Does I have the wrong configuration of Spark? I tried several permutation of my spark-defaults.conf. I tried 3 Executor per node and more, I changed the memory size and so on. But everytime it ends up in the same exception.

Maybe someone has an idea for this problem?

Kind regards

Sedir Mohammed

spark.executor.instances 8

spark.executor.cores 10

val graph = Graph(vertices, edges).partitionBy(EdgePartition2D, 4)
^^^

There is no point in having 8 executors with 10 cores each available if you make only 4 partitions. Bear in mind that all partitions on an executor together must fit in memory in order to avoid GC thrashing. Try using a larger number of partitions such that 10 partitions easily fit in memory, maybe something like a few hundred MB of input data per partition.
Also, please check that the 10GB of RAM are actually available on every worker node and 16GB on the machine where the driver is executed - if there is no RAM available on some of your workers, you can reduce the number of cores and the amount of memory in your Spark configuration.