如何使用GraphX的Java API获取图形的连接组件的列表

Question

我对Spark和GraphX相当陌生，我正在尝试了解如何使用GraphX的Java API执行以下操作。我希望生成具有以下签名的方法：

private <List<Graph<VD, ED>> computeConnectedComponents(Graph<VD, ED> graph){}

其中，给定一个只有正度节点的图，但连接分量未知，它应该返回一个图列表（顺序无关紧要），每个图都是连接的。

我知道 GraphOps.connectedComponents（） 和 ConnectedComponents.run（），但我正在努力理解返回值。文档将它们列为返回图形的图形

基本上，我想知道我能做些什么来从 connectedComponents 的返回值和我的初始图中派生这个图列表。




Answer 1

以下代码是用 scala 编写的，但应该演示这个想法。

返回的图形将包含所有顶点，但每个顶点的属性将替换为 VertexId（实际上只是一个 Long），它可以解释为顶点所属的连接组件的 id。它也是属于该连接组件的“最低顶点 id”。

import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD
val vertexArray = Array(
  (1L, ("A", 28)),
  (2L, ("B", 27)),
  (3L, ("C", 65)),
  (4L, ("D", 42)),
  (5L, ("E", 55)),
  (6L, ("F", 50)),
  (7L, ("G", 53)),
  (8L, ("H", 66))
  )

// Vertices 1 - 6 are connected, 7 and 8 are connected.
val edgeArray = Array(
  Edge(2L, 1L, 7),
  Edge(2L, 4L, 2),
  Edge(3L, 2L, 4),
  Edge(3L, 6L, 3),
  Edge(4L, 1L, 1),
  Edge(5L, 2L, 2),
  Edge(5L, 3L, 8),
  Edge(5L, 6L, 3),
  Edge(7L, 8L, 3)
  )

val vertexRDD: RDD[(Long, (String, Int))] = sc.parallelize(vertexArray)
val edgeRDD: RDD[Edge[Int]] = sc.parallelize(edgeArray)
val graph: Graph[(String, Int), Int] = Graph(vertexRDD, edgeRDD)

val cc = graph.connectedComponents().vertices.collectAsMap()
cc.foreach {
  case (vertexId, clusterId) =>
    println(s"Vertex $vertexId belongs to cluster $clusterId")
}

输出：

Vertex 8 belongs to cluster 7
Vertex 2 belongs to cluster 1
Vertex 5 belongs to cluster 1
Vertex 4 belongs to cluster 1
Vertex 7 belongs to cluster 7
Vertex 1 belongs to cluster 1
Vertex 3 belongs to cluster 1
Vertex 6 belongs to cluster 1