使用 Pyspark 的最短路径
输入数据可以解释为一个图表，其中包含之间的连接currentnode and childnode。那么问题是根节点和所有叶子节点之间的最短路径是多少并被称为单源最短路径 https://en.wikipedia.org/wiki/Shortest_path_problem#Single-source_shortest_paths.
火花有Graphx https://spark.apache.org/docs/2.4.6/graphx-programming-guide.html处理图的并行计算。不幸的是，GraphX不提供Python API（更多细节可以找到here https://stackoverflow.com/q/23302270/2129801）。一个支持Python的图形库是图框 https://graphframes.github.io/graphframes/docs/_site/index.html。 GraphFrames 使用 GraphX 的一部分。
GraphX 和 GraphFrames 都提供了 sssp 的解决方案。再次不幸的是，两种实现都仅返回最短路径的长度，而不返回路径本身（GraphX https://spark.apache.org/docs/2.4.6/api/scala/index.html#org.apache.spark.graphx.lib.ShortestPaths%24 and 图框 https://graphframes.github.io/graphframes/docs/_site/user-guide.html#shortest-paths). But 这个答案 https://stackoverflow.com/a/27109143/2129801为 GraphX 和 Scala 提供算法的实现，同时返回路径。所有三个解决方案都使用Pregel https://dzone.com/articles/distributed-graphs-processing-with-pregel.
将上述答案翻译为 GraphFrames/Python：
1. 数据准备
为所有节点提供唯一的 ID 并更改列名称，使其适合所描述的名称here https://graphframes.github.io/graphframes/docs/_site/api/python/graphframes.html#graphframes.GraphFrame
import pyspark.sql.functions as F
df = ...
vertices = df.select("currentnode").withColumnRenamed("currentnode", "node").union(df.select("childnode")).distinct().withColumn("id", F.monotonically_increasing_id()).cache()
edges = df.join(vertices, df.currentnode == vertices.node).drop(F.col("node")).withColumnRenamed("id", "src")\
.join(vertices, df.childnode== vertices.node).drop(F.col("node")).withColumnRenamed("id", "dst").cache()
Nodes Edges
+------+------------+ +-----------+---------+------------+------------+
| node| id| |currentnode|childnode| src| dst|
+------+------------+ +-----------+---------+------------+------------+
| leaf2| 17179869184| | child1| leaf4| 25769803776|249108103168|
|child1| 25769803776| | child1| child3| 25769803776| 68719476736|
|child3| 68719476736| | child1| leaf2| 25769803776| 17179869184|
| leaf6|103079215104| | child3| leaf6| 68719476736|103079215104|
| root|171798691840| | child3| leaf5| 68719476736|214748364800|
| leaf5|214748364800| | root| child1|171798691840| 25769803776|
| leaf4|249108103168| +-----------+---------+------------+------------+
+------+------------+
2. 创建GraphFrame
from graphframes import GraphFrame
graph = GraphFrame(vertices, edges)
3. 创建将构成 Pregel 算法的各个部分的 UDF
The message type:
from pyspark.sql.types import *
vertColSchema = StructType()\
.add("dist", DoubleType())\
.add("node", StringType())\
.add("path", ArrayType(StringType(), True))
顶点程序：
def vertexProgram(vd, msg):
if msg == None or vd.__getitem__(0) < msg.__getitem__(0):
return (vd.__getitem__(0), vd.__getitem__(1), vd.__getitem__(2))
else:
return (msg.__getitem__(0), vd.__getitem__(1), msg.__getitem__(2))
vertexProgramUdf = F.udf(vertexProgram, vertColSchema)
传出消息：
def sendMsgToDst(src, dst):
srcDist = src.__getitem__(0)
dstDist = dst.__getitem__(0)
if srcDist < (dstDist - 1):
return (srcDist + 1, src.__getitem__(1), src.__getitem__(2) + [dst.__getitem__(1)])
else:
return None
sendMsgToDstUdf = F.udf(sendMsgToDst, vertColSchema)
消息聚合：
def aggMsgs(agg):
shortest_dist = sorted(agg, key=lambda tup: tup[1])[0]
return (shortest_dist.__getitem__(0), shortest_dist.__getitem__(1), shortest_dist.__getitem__(2))
aggMsgsUdf = F.udf(aggMsgs, vertColSchema)
4. 组合零件
from graphframes.lib import Pregel
result = graph.pregel.withVertexColumn(colName = "vertCol", \
initialExpr = F.when(F.col("node")==(F.lit("root")), F.struct(F.lit(0.0), F.col("node"), F.array(F.col("node")))) \
.otherwise(F.struct(F.lit(float("inf")), F.col("node"), F.array(F.lit("")))).cast(vertColSchema), \
updateAfterAggMsgsExpr = vertexProgramUdf(F.col("vertCol"), Pregel.msg())) \
.sendMsgToDst(sendMsgToDstUdf(F.col("src.vertCol"), Pregel.dst("vertCol"))) \
.aggMsgs(aggMsgsUdf(F.collect_list(Pregel.msg()))) \
.setMaxIter(10) \
.setCheckpointInterval(2) \
.run()
result.select("vertCol.path").show(truncate=False)
Remarks:
-
maxIter应设置为至少与最长路径一样大的值。如果该值较高，结果将保持不变，但计算时间会变长。如果该值太小，结果中将丢失较长的路径。当前版本的 GraphFrames (0.8.0) 不支持在不再发送新消息时停止循环。
-
checkpointInterval应设置为小于maxIter。实际值取决于数据和可用硬件。当发生 OutOfMemory 异常或 Spark 会话挂起一段时间时，该值可能会减小。
最终结果是一个包含内容的常规数据框
+-----------------------------+
|path |
+-----------------------------+
|[root, child1] |
|[root, child1, leaf4] |
|[root, child1, child3] |
|[root] |
|[root, child1, child3, leaf6]|
|[root, child1, child3, leaf5]|
|[root, child1, leaf2] |
+-----------------------------+
如果需要，可以在这里过滤掉非叶节点。
