【发布时间】：2017-01-30 16:06:45
【问题描述】：
我有一个函数，它采用节点的邻居，对于邻居我使用广播变量和节点本身的 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
}
【问题讨论】：
-
您是在集群还是本地启动程序？如果是本地主机，您是否指定了要使用的核心数，例如：
--master=local[8]。或者，您的数据集有多少个分区？如果它只有一个分区，则使用单个核心。 -
是的，我使用其他程序使用了更多资源。对于分区，我保留了从边缘列表文件加载图形时的默认值，但我想到了这一点，我使用了 coalesce我有 10 个 8 个内核，我应该使用更多还是我做错了；
-
你能提供一些代码吗？
-
感谢您的代码。奇怪的是，您不使用 GraphX 中的任何方法来处理图形。诸如计算最短路径之类的事情应该使用 Pregel API 来完成。此外，在您的第一个指令中，您似乎正在将 graoh 的整个邻接部分收集到驱动程序上。这是火花的一大“代码气味”。我知道这并不能真正回答问题，但您可能希望更改解决问题的方法以更好地利用 Spark 的分布式特性。
-
Graphx 提供 Pregel API 来计算图节点的 SSSP，这对我来说是相同的，但是你不能为每个节点并行执行此操作，就像使用它自己的图一样节点并行。我还没有想到更好的东西。起初我虽然相同，因为找到所有邻居几乎和图一样大，这不好但我不知道如何并行。