从大型 JSON 文件创建树状结构的最有效方法

Posted 2023-03-16

【中文标题】从大型 JSON 文件创建树状结构的最有效方法

【英文标题】：Most efficient way to create a tree like structure from a large JSON file

【发布时间】：2022-01-17 10:52:27

【问题描述】：

我有一个大的 JSON 文件，如下所示：

[(3, (2, 'Child')), (2, (1, 'Parent')), (1, (None, 'Root'))]

其中每个元素的键是该元素的唯一索引，值对中的第一个元素表示其父元素的索引。

现在，最终目标是将此 JSON 文件转换为以下内容：

[(3, (2, 'Child Parent Root')), (2, (1, 'Parent Root')), (1, (None, 'Root'))]

其中每个项目的值对中的第二个元素将被修改，以便它具有所有值的串联，直到其根祖先。

没有。级别数不固定，最多可达 256。我知道我可以通过创建树 DS 并遍历它来解决这个问题，但问题是 JSON 文件很大（列表中几乎有 1.8 亿个项目）。

知道如何有效地实现这一目标吗？涉及 Apache Spark 的建议也可以。

【问题讨论】：

我们可以假设父级的索引必须小于子级吗？

是的，可以假设。

您呈现的 JSON 文件不是 JSON。 JSON 不知道元组的语法，也不将单引号视为字符串分隔符。

【参考方案1】：

您可以使用广度优先搜索来查找所有祖先元素链：

from collections import deque, defaultdict
d, d1 = [(3, (2, 'Child')), (2, (1, 'Parent')), (1, (None, 'Root'))], defaultdict(list)
for a, (b, c) in d:
   d1[b].append((a, c))

q, r = deque([(1, d1[None][0][1])]), 
while q:
   r[n[0]] = (n:=q.popleft())[1]
   q.extend([(a, b+' '+n[1]) for a, b in d1[n[0]]])

现在，r 存储每个元素的祖先值：

1: 'Root', 2: 'Parent Root', 3: 'Child Parent Root'

然后，使用列表推导更新d：

result = [(a, (b, r[a])) for a, (b, _) in d]

输出：

[(3, (2, 'Child Parent Root')), (2, (1, 'Parent Root')), (1, (None, 'Root'))]

BFS 等迭代方法将消除在非常大的图上运行 DFS 时可能出现RecursionError 的可能性。

【讨论】：

感谢您的回答。这是正确的，但对于 数组中的约 1.8 亿个项目似乎不够有效。在处理给定的“while”循环时出现“Killed”错误。

【参考方案2】：

在 Spark 中，这被视为 Graph 问题并使用 Vertex Centric Programming 解决。不幸的是，GraphX 没有兼容 python 的 API。

另一种选择是使用graphframes。

我在这里包含了一个使用模仿 Vertex Centric Programming 但不使用任何库的连接的逻辑。前提是您可以将拥有的数据表示形式转换为具有id、parent_id 和name 列的数据框。

from pyspark.sql import functions as F
from pyspark.sql.functions import col as c
from pyspark.sql import SparkSession

spark = SparkSession.builder.master("local[1]").appName("nsankaranara").getOrCreate() 

data = [(6, 3, "Child_3", ),
        (5, 2, "Child_2", ),
        (4, 2, "Child_1", ),
        (3, 1, "Parent_2", ), 
        (2, 1, "Parent_1", ), 
        (1, None, "Root"), ]


df = spark.createDataFrame(data, ("id", "parent_id", "name", ))

df = (df.withColumn("mapped_parent_id", c("parent_id"))
        .withColumn("visited", F.lit(False)))

start_nodes = df.filter(c("mapped_parent_id").isNull())

# Controls how deep we want to traverse
max_iter = 256
iter_counter = 0

# Iteratively identify the next child and add your name to them
while iter_counter < max_iter:
    iter_counter += 1
    df = (df.alias("a").join(start_nodes.alias("b"), 
                            ((c("a.parent_id") == c("b.id")) | (c("a.id") == c("b.id"))), how="left_outer"))
    df = (df.select(c("a.id"), 
                  c("a.parent_id"),
                  (F.when((c("b.id").isNotNull() & (c("a.id") != c("b.id"))), F.lit(None)).otherwise(c("a.mapped_parent_id"))).alias("mapped_parent_id"),
                  F.when((c("a.id") != c("b.id")), F.concat_ws(" ", c("a.name"), c("b.name"))).otherwise(c("a.name")).alias("name"),
                  (F.when(c("a.id") == c("b.id"), F.lit(True)).otherwise(c("a.visited"))).alias("visited")
                  )).cache()
    
    start_nodes = df.filter(((c("mapped_parent_id").isNull()) & (c("visited") == False)))
    if start_nodes.count() == 0:
        # signifies that all nodes have been visited
        break

df.select("id", "parent_id", "name").show(truncate = False)

输出

+---+---------+---------------------+
|id |parent_id|name                 |
+---+---------+---------------------+
|6  |3        |Child_3 Parent_2 Root|
|5  |2        |Child_2 Parent_1 Root|
|4  |2        |Child_1 Parent_1 Root|
|3  |1        |Parent_2 Root        |
|2  |1        |Parent_1 Root        |
|1  |null     |Root                 |
+---+---------+---------------------+

【讨论】：

我在执行此实现时反复收到此错误。 “org.apache.spark.SparkException：无法在 300 秒内执行广播。您可以通过 spark.sql.broadcastTimeout 增加广播的超时时间或通过将 spark.sql.autoBroadcastJoinThreshold 设置为 -1 来禁用广播加入”...我有尝试禁用广播加入和增加广播超时，但没有任何效果。知道如何解决这个问题吗？

禁用autoBroadcastJoin 应该可以工作。然而，另一种选择是在select 之后缓存df。这将打破血统链并减少广播所需的时间。

我已编辑答案以在选择后包含cache。