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を使用してさらに多くのパーティションを取得できます。あなたは​​3210をmentionnedだけ、デフォルトでは、パーティションの数を減らすこと

、この質問を参照してください。Spark Coalesce More Partitions