Styjun

I am working with Spark GraphX. I am building a graph from a file (around 620 mb, 50K vertices and almost 50 millions of edges). I am using a spark cluster with: 4 workers, each one with 8 cores and 13.4g of ram, 1 driver with the same specs. When I submit my .jar to the cluster, randomly one of the workers loads all the data on it. All the task needed for the computing are requested to that worker. While the computing the remaining three are without doing nothing. I have try everything and i do not found nothing that can force to compute in all of the workers.

When Spark build the graph and I look for the number of partitions of the RDD of vertices say 5, but if I repartition that RDD for example with 32 (number of cores in total) Spark load the data in every worker but gets slow the computation.

Im launching the spark submit by this way:

spark-submit --master spark://172.30.200.20:7077 --driver-memory 12g --executor-memory 12g --class interscore.InterScore /root/interscore/interscore.jar hdfs://172.30.200.20:9000/user/hadoop/interscore/network.dat hdfs://172.30.200.20:9000/user/hadoop/interscore/community.dat 111

The code is here:

object InterScore extends App
 val sparkConf = new SparkConf().setAppName("Big-InterScore")
 val sc = new SparkContext(sparkConf)

 val t0 = System.currentTimeMillis
 runInterScore(args(0), args(1), args(2))
 println("Running time " + (System.currentTimeMillis - t0).toDouble / 1000)

 sc.stop()

 def runInterScore(netPath:String, communitiesPath:String, outputPath:String) = 
 val communities = sc.textFile(communitiesPath).map(x => 
 val a = x.split('t')
 (a(0).toLong, a(1).toInt)
 ).cache

 val graph = GraphLoader.edgeListFile(sc, netPath, true)
 .partitionBy(PartitionStrategy.RandomVertexCut)
 .groupEdges(_ + _)
 .joinVertices(communities)((_, _, c) => c)
 .cache

 val lvalues = graph.aggregateMessages[Double](
 m => 
 m.sendToDst(if (m.srcAttr != m.dstAttr) 1 else 0)
 m.sendToSrc(if (m.srcAttr != m.dstAttr) 1 else 0)
 , _ + _)

 val communitiesIndices = communities.map(x => x._2).distinct.collect
 val verticesWithLValue = graph.vertices.repartition(32).join(lvalues).cache
 println("K = " + communitiesIndices.size)
 graph.unpersist()
 graph.vertices.unpersist()
 communitiesIndices.foreach(c => 
 //COMPUTE c
 
 )
 
}