Spark,Graphx程序不使用cpu和内存

Question

我有一个函数，它获取一个节点的邻居，对于邻居，我使用广播变量和节点本身的id，它计算该节点的贴近度中心度。我用该函数的结果映射图的每个节点。当我打开任务管理器时，cpu根本没有被利用，就像它没有并行工作一样，内存也是一样，但是每个节点都是并行执行功能的，而且数据很大，完成起来需要时间，并不是不需要资源。非常感谢大家的帮助，谢谢。对于加载图形，我使用val graph=graphloader.edgelistfile(sc,path).cache

object ClosenessCentrality {

  case class Vertex(id: VertexId)

  def run(graph: Graph[Int, Float],sc: SparkContext): Unit = {
    //Have to reverse edges and make graph undirected because is bipartite
    val neighbors = CollectNeighbors.collectWeightedNeighbors(graph).collectAsMap()
    val bNeighbors = sc.broadcast(neighbors)

    val result = graph.vertices.map(f => shortestPaths(f._1,bNeighbors.value))
    //result.coalesce(1)
    result.count()

  }

  def shortestPaths(source: VertexId,  neighbors: Map[VertexId, Map[VertexId, Float]]): Double ={
    val predecessors = new mutable.HashMap[VertexId, ListBuffer[VertexId]]()
    val distances = new mutable.HashMap[VertexId, Double]()
    val q = new FibonacciHeap[Vertex]
    val nodes = new mutable.HashMap[VertexId, FibonacciHeap.Node[Vertex]]()

    distances.put(source, 0)

    for (w <- neighbors) {
      if (w._1 != source)
        distances.put(w._1, Int.MaxValue)

      predecessors.put(w._1, ListBuffer[VertexId]())
      val node = q.insert(Vertex(w._1), distances(w._1))
      nodes.put(w._1, node)
    }

    while (!q.isEmpty) {
      val u = q.minNode
      val node = u.data.id
      q.removeMin()
      //discover paths
      //println("Current node is:"+node+" "+neighbors(node).size)
      for (w <- neighbors(node).keys) {
        //print("Neighbor is"+w)
        val alt = distances(node) + neighbors(node)(w)
//        if (distances(w) > alt) {
//          distances(w) = alt
//          q.decreaseKey(nodes(w), alt)
//        }
//        if (distances(w) == alt)
//          predecessors(w).+=(node)
         if(alt< distances(w)){
           distances(w) = alt
           predecessors(w).+=(node)
           q.decreaseKey(nodes(w), alt)
         }

      }//For
    }
    val sum = distances.values.sum
    sum
  }

Answer 1

为了回答您最初的问题，我怀疑您的RDD只有一个分区，因此使用一个内核进行处理。

edgelistfile方法有一个参数来指定所需的最小分区数。此外，还可以使用repartition获得更多分区。

您提到了Coalesce，但这在默认情况下只会减少分区的数量，请参见以下问题：Spark Coalesce More分区