Scala получает список смежности каждой вершины из списка исходного узла и конечного узла

Question

Я довольно новичок в Scala и GraphX, так что это может быть тривиальным вопросом. Я пытаюсь изменить направление каждого ребра графа, а затем получить список смежности каждой вершины нового перевернутого графа. Ввод осуществляется в форме "FromNodeId \t ToNodeId"

0 1
0 2
1 2
1 3

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

object Problem2{
    def main(args: Array[String]){
    val inputFile:String = args(0)
    val outputFolder = args(1)
    val conf = new SparkConf().setAppName("Problem2").setMaster("local")
    val sc = new SparkContext(conf)
    val input = sc.textFile(inputFile)
    val graph = GraphLoader.edgeListFile(sc,inputFile)
    val newGraph = graph.reverse.edges
}
}

Выход имеет форму

Edge(1,0,1)
Edge(3,1,1)
Edge(2,1,1)
Edge(2,0,1)

Мои вопросы: 1. Есть ли более эффективный подход к проблеме 2. Как я могу приступить к построению списка смежности для каждой вершины отсюда?

Answer 1

Ответ на оба вопроса - это collectNeighbors и collectNeighborIds.

Для каждой вершины он вычисляет список соседей. Обратите внимание на параметр EdgeDirection.

val vertices: RDD[(VertexId, String)] =  
    sc.parallelize(Array((1L,""), (2L,""), (4L,""), (6L,"")))


val edges: RDD[Edge[String]] = 
    sc.parallelize(Array(
        Edge(1L, 2L, ""),
        Edge(1L, 4L, ""),
        Edge(1L, 6L, "")))
val inputGraph = Graph(vertices, edges)

val verticesWithSuccessors: VertexRDD[Array[VertexId]] = 
    inputGraph.ops.collectNeighborIds(EdgeDirection.Out)

Так как граф является неизменным, вы должны построить новый. Те же ребра, новые вершины

val successorGraph = Graph(verticesWithSuccessors, edges)

И вот частичный результат:

val res = successorGraph.vertices.collect()
res: Array[(org.apache.spark.graphx.VertexId,
         Array[org.apache.spark.graphx.VertexId])] = 
  Array((4,Array()), (1,Array(2, 4, 6)), (6,Array()), (2,Array()))