Apache Spark вычисление кратчайшего пути

Question

Я пытаюсь вычислить кратчайший путь в большой сети от заданного источника к заданной цели на основе весов без использования Apache Spark. Поскольку весь мой другой код написан на python, я не хочу ничего менять. Это должно быть как-то возможно, не так ли? Поскольку я совершенно новичок в Spark, возможно, я не понимаю, как я могу решить эту проблему.

Может быть, кто-то может мне помочь? Заранее спасибо!

Что я пробовал до сих пор:

• создание списка вершин и ребер
• используя GraphFrame() для создания графика
• использование метода кратчайшего пути GraphFrames для вычисления кратчайшего пути

Пока все хорошо (не совсем). Проблема с методом кратчайшего пути GraphFrames состоит в том, что он вычисляет кратчайший путь от каждого узла до заданного набора узлов, который работает для небольших графов, но для огромных сетей требуется много времени. Много "ненужных" вычислений выполняется потому, что учитываются все узлы. Мне просто нужно получить кратчайший путь от одного узла к другому.

Я искал в интернете и обнаружил, что в библиотеке Spark graphx есть такая функция, которую я ищу, но, к сожалению, она доступна только для Scala...

Может быть, я просто могу использовать RDDS для вычисления кратчайшего пути на основе весов? Или есть кратчайшая реализация пути для pyspark, которую я не смог найти? Не могу поверить, что для pyspark не реализован алгоритм кратчайшего пути.

    vertices_rdd = vertices_rdd3.zipWithIndex()
    # vertices_rdd.take(3): 
    # [((552897.813699282, 4164322.19502139), 0), ((583743.487097408, 4158379.86761575), 1), ((585964.589845657, 4158443.96863072), 2)]

    edges_rdd = edges_rdd1.flatMap(lambda x: x)
    # edges_rdd.take(3): 
    # [(62734, 107857, 102.19468251940246, '8'), (107857, 62734, 102.19468251940246, '8'), (79903, 191109, 21.81675476329727, '13')]

    spark = SparkSession(sc)

    vertices_df = vertices_rdd.toDF(["coordinate","id"])
    edges_df = edges_rdd.toDF(["src", "dst", "distance", "streetclass"])

    vertices_df.show()
    #+--------------------+---+
    #|          coordinate| id|
    #+--------------------+---+
    #|[552897.813699282...|  0|
    #|[583743.487097408...|  1|
    #|[585964.589845657...|  2|
    #|[588646.795215483...|  3|
    #|[582405.137425844...|  4|
    #|[582823.612980657...|  5|
    #...

    edges_df.show()
    #+------+------+------------------+-----------+
    #|   src|   dst|          distance|streetclass|
    #+------+------+------------------+-----------+
    #| 62734|107857|102.19468251940246|          8|
    #|107857| 62734|102.19468251940246|          8|
    #| 79903|191109| 21.81675476329727|         13|
    #|191109| 79903| 21.81675476329727|         13|
    #| 60790| 66205|19.362434806339824|         13|
    #... 

    from graphframes import *
    g = GraphFrame(vertices_df, edges_df)

    results = g.shortestPaths(landmarks=["0"])
    results.select("id", "distances").show()
    #+---+-----------+
    #| id|  distances|
    #+---+-----------+
    #|  0|Map(0 -> 0)|
    #|  7|Map(0 -> 1)|
    #|  6|      Map()|
    #|  9|      Map()|
    #|  5|      Map()|
    #|  1|      Map()|
    #|  3|      Map()|
    #|  8|      Map()|
    #...