如何使用 Scala 在 Spark GraphX 中查找特定节点与其邻居之间的交集

Question

我是 spark graphx 的新手，并尝试分布式计算特定节点之间的交集，例如 ID = 1 的节点与其在 spark GraphX 中的邻居。

我已经使用 GraphLoader.edgeListFile(sc,"Path") 加载了边缘列表。然后我用 collectNeighborIds 找到节点 id = 1 的邻居 ID，并在其上执行映射函数以查找每个邻居邻居并计算与所选节点（ID = 1 的节点）邻居的交集。 这是代码。

val graph = GraphLoader.edgeListFile(sc,path to edgelist)
val node_collect_neighborsId1 = graph.collectNeighborIds(EdgeDirection.Either).filter(x=> x._1 == 1)

val node1_neighbors_ID=node_collect_neighborsId1.flatMap(x=> x._2)

def compute_intersection (vertex :VertexId) = {


  var node2_collect_neighborsId: RDD[(VertexId, Array[VertexId])] = graph.collectNeighborIds(EdgeDirection.Either).filter(x=> x._1 == vertex)

  var node2_neighbors_ID=node2_collect_neighborsId.flatMap(x=> x._2)

  var intersect_two_node = node1_neighbors_ID.intersection(node2_neighbors_ID)

  (vertex, intersect)

}

val result = node1_neighbors_ID.map(compute_intersection)

我希望在最后，结果变量应该包含包含顶点 id 的行，该顶点 id 是邻居的 id 和我们称之为交集的两组节点邻居之间的公共节点。但我不能打印它们，看看里面有什么。 请帮我解决计算交集和打印结果的问题

Answer 1

您无法构建RDD[RDD[T]] 类型的结果。因此，您不应该为 map 内的每个邻居计算交集。

您可以使用aggregateMessages 计算所有目标邻居的交集：

def computeIntersection[VD, ED](graph: Graph[VD, ED], targetVertexId: Long): VertexRDD[List[Long]] = {
  //mark the target's neighbors
  val verticesWithTargetNeighborFlag = graph.aggregateMessages[Boolean](
    triplet => {
      if(triplet.srcId == targetVertexId && triplet.dstId != targetVertexId) {
        triplet.sendToDst(true)
      } else if(triplet.dstId == targetVertexId && triplet.dstId != targetVertexId) {
        triplet.sendToSrc(true)
      }
    },
    (msg1, msg2) => msg1 || msg2,
    TripletFields.None
  )
  val graphWithTargetNeighborFlag = Graph(verticesWithTargetNeighborFlag, edges)
  //collect intersection vertices for each target's neighbor
  val verticesWithIntersection = graphWithTargetNeighborFlag.aggregateMessages[List[Long]](
    triplet => if (triplet.srcAttr && triplet.dstAttr) { //both are target's neighbors
      triplet.sendToDst(List(triplet.srcId))
      triplet.sendToSrc(List(triplet.dstId))
    },
    (msg1, msg2) => msg1 ::: msg2,
    TripletFields.All
  )
  verticesWithIntersection
}

你可以使用collect打印RDD元素：

rdd.collect().foreach(println)