Как реализовать письменный код в aggregateMessages в Pregel API в Spark?

Question

Я реализовал вычисление сходства между узлами графа в aggregateMessages. при этом пересечение или общие соседи между двумя узлами вычисляются и отправляются в оба. сообщение является двойным числом. каждый узел получает его и суммирует, чтобы вычислить сумму подобия для себя. сходство известно как сходство с Джакардом.

у меня есть график, структура которого выглядит следующим образом:

(vertexID, List[neighbors ID])
(vertexID, List[neighbors ID])
(vertexID, List[neighbors ID])
...
(vertexID, List[neighbors ID])

например:

(1, List[2,6,8,9])
(2, List[12,8,7,9])
(3, List[4,22,33,16])
...

код, написанный в aggregateMessages как показано ниже:

    val nodes_similarity_sum: RDD[(VertexId, Double)] = graph.aggregateMessages[Double](
      sendMsg = triplet => {
        val srcNeighbor = triplet.srcAttr
        val dstNeighbor = triplet.dstAttr

        val temp_intersect = srcNeighbor.intersect(dstNeighbor).length
        val temp_union = srcNeighbor.union(dstNeighbor).length
        val similarity =(((temp_intersect.toFloat) / (temp_union.toFloat)))

        triplet.sendToDst(similarity)
        triplet.sendToSrc(similarity)
      },
      mergeMsg = (x, y) => x + y
    )

Я считаю, что если я реализую это в Pregel, это будет более оптимизировано и быстрее. но у меня есть проблемы в реализации.

Может ли кто-нибудь реализовать его в Pregel? это было бы так полезно и интересно !!!