Spark：为层次结构DataFrame的每个节点构建递归树路径

Question

考虑一棵树及其DataFrame表示（左表）：

0             ┌───────┬───────┐           ┌───────┬───────┐
├──1          │   id  │ parent│           │   id  │ path  │
│  ├──2       ├───────┼───────┤           ├───────┼───────┤
│  └──3       │   5   │   0   │           │   5   │0/5    │
│     └──4    ├───────┼───────┤           ├───────┼───────┤
└──5          │   4   │   3   │           │   4   │0/1/3/4│
              ├───────┼───────┤     =>    ├───────┼───────┤
              │   3   │   1   │           │   3   │0/1/3  │
              ├───────┼───────┤           ├───────┼───────┤
              │   2   │   1   │           │   2   │0/1/2  │
              ├───────┼───────┤           ├───────┼───────┤
              │   1   │   0   │           │   1   │0/1    │
              ├───────┼───────┤           ├───────┼───────┤
              │   0   │ null  │           │   0   │0      │
              └───────┴───────┘           └───────┴───────┘

如何最高效地获取树的每个节点（右表）从根节点开始的路径？

所有可能的方法都可以使用：SQL查询、DataFrame方法、GraphX 等。

注意：经典 SQL 解决方案中的递归连接对于 Spark DataFrames 不起作用。

- Oleg Mikhailov

---

我怀疑GraphX可能是正确的选择，但我怀疑它的效率不会很高。 - Shaido

是的，看起来这个任务可以在不初始化图形的情况下解决。 - Oleg Mikhailov

@OlegMikhailov，RDD的mapPartitions怎么样？ - Sai

@Sai，只要方法有效，它们都是好的。 - Oleg Mikhailov

@OlegMikhailov，为什么你说“经典的SQL解决方案使用递归连接在Spark DataFrames中不起作用。”？我认为在Spark中，大型表格连接（在这种情况下是自身连接）很快。 - travelingbones

Answer 1

这似乎是一个Spark Graph API任务。您可以查看Graphframes Spark包。这是一个提供高级API的包，可在GraphX核心上构建图表（与传统的基于RDD的Spark Dataframes使用相同）。借助此功能，您可以使用数据帧构建图形。

请参阅此链接：https://mapr.com/blog/analyzing-flight-delays-with-apache-spark-graphframes-and-mapr-db/ 它展示了航班数据的用例。如果您查看“广度优先搜索图算法”部分，您将看到一种确切地实现您想要的算法：在两个顶点之间找到路径（给定maxPathLength参数）。

按照您的Spark版本运行pyspark并附带graphframes依赖项。

pyspark --packages graphframes:graphframes:0.6.0-spark2.3-s_2.11

构建你的数据框：

df = sc.parallelize([{"id": 5, "parent": 0}, {"id": 4, "parent": 3}, {"id": 3, "parent": 1}, {"id": 2, "parent": 1}, {"id": 1, "parent": 0}, {"id": 0, "parent": None}]).toDF()

创建图表：

df_vertices = df.selectExpr("id")
df_edges = df.withColumnRenamed("id", "dst").withColumnRenamed("parent", "src")

from graphframes import GraphFrame
graph  = GraphFrame(df_vertices, df_edges)

可视化路径（例如从0到4）：

graph.bfs(fromExpr="id = 0",toExpr="id = 4", maxPathLength=10).show(2)

结果：

+----+------+---+------+---+------+---+
|from|    e0| v1|    e1| v2|    e2| to|
+----+------+---+------+---+------+---+
| [0]|[1, 0]|[1]|[3, 1]|[3]|[4, 3]|[4]|
+----+------+---+------+---+------+---+

- Igor Uchôa