使用 Pyspark 从关系数据集构建层次结构

Question

我是 Python 新手，一直致力于从关系数据集构建层次结构。
如果有人知道如何进行此操作，那将有巨大的帮助。

我有一个关系数据集，其中包含如下数据

_currentnode,  childnode_  
 root,         child1  
 child1,       leaf2  
 child1,       child3  
 child1,       leaf4  
 child3,       leaf5  
 child3,       leaf6  

很快。我正在寻找一些 python 或 pyspark 代码来
构建如下所示的层次结构数据框

_level1, level2,  level3,  level4_  
root,    child1,  leaf2,   null  
root,    child1,  child3,  leaf5  
root,    child1,  child3,  leaf6  
root,    child1,  leaf4,   null  

这些数据是字母数字，是一个巨大的数据集[约 5000 万条记录]。
此外，层次结构的根是已知的，并且可以在代码中硬连线。
因此，在上面的示例中，层次结构的根是“root”。

Answer 1

使用 Pyspark 的最短路径

currentnode输入数据可以解释为具有和之间连接的图表childnode。那么问题就是根节点到所有叶子节点之间的最短路径是什么，称为单源最短路径。

Spark 有Graphx来处理图的并行计算。不幸的是，GraphX 不提供 Python API（更多详细信息可以在这里找到）。支持 Python 的图形库是GraphFrames。GraphFrames 使用 GraphX 的一部分。

GraphX 和 GraphFrames 都提供了 sssp 的解决方案。不幸的是，这两种实现都只返回最短路径的长度，而不返回路径本身（GraphX和GraphFrames）。但这个答案提供了 GraphX 和 Scala 算法的实现，它也返回路径。所有三种解决方案均使用Pregel。

将上述答案翻译为 GraphFrames/Python：

1. 数据准备

为所有节点提供唯一 ID 并更改列名称，使其适合此处描述的名称

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

消息类型：

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)   

评论：

• 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]        |
+-----------------------------+

如果需要，可以在这里过滤掉非叶节点。