使用 PySpark 解析具有大量唯一键（不是对象列表）的 JSON 对象

Question

我目前正在处理 JSON 文件中的以下源数据：

{
    "unique_key_1": {
        "some_value_1": 1,
        "some_value_2": 2
    },
    "unique_key_2": {
        "some_value_1": 2,
        "some_value_2": 3
    }
    "unique_key_3": {
        "some_value_1": 2,
        "some_value_2": 1
    }
    ...
}

请注意，源数据实际上是一个大型字典，其中包含许多唯一键。它不是字典列表。我有很多像这样的大型 JSON 文件，我想使用 PySpark 将它们解析为以下 DataFrame 结构：

key          | some_value_1 | some_value_2
-------------------------------------------
unique_key_1 |            1 |            2
unique_key_2 |            2 |            3
unique_key_3 |            2 |            1

如果我正在处理小文件，我可以简单地使用类似于以下的代码来解析它：

[{**{"key": k}, **v} for (k, v) in source_dict.items()] 

然后，我将在此列表上创建一个 Spark DataFrame 并继续我需要执行的其余操作。

我的问题是我不太清楚如何将这样的大型 JSON 对象解析为 DataFrame。当我使用SPARK.read.json("source_dict.json") 时，我得到一个DataFrame，其中每一个唯一的key 值（可以预见地）作为一列读入。请注意，真实的数据文件可能有成千上万个这样的键。

我对 Spark 世界还很陌生，我似乎找不到完成这项任务的方法。这似乎是一个支点或类似的东西会有所帮助。有没有人有任何解决方案或可能的解决方案的指针？谢谢，不胜感激！

Answer 1

使用 flatmap 你可以编写一个函数来进行转换

def f(row):
l = []
d = row.asDict()
for k in d.keys():
    l.append(Row(k, d[k][0], d[k][1]))
return Row(*l)

rdd = df.rdd.flatMap(f)
spark.createDataFrame(rdd).show()


+------------+---+---+
|          _1| _2| _3|
+------------+---+---+
|unique_key_1|  1|  2|
|unique_key_2|  2|  3|
|unique_key_3|  2|  1|
+------------+---+---+

有关更多信息，您可以查看link

Answer 2

将键放入单独列的最简单方法是在将数据读入 Spark 之前重构 json。如果 JSON 的结构如下：

[
    {"key":"unique_key_1",
        "some_value_1": 1,
        "some_value_2": 2
    },
    {"key":"unique_key_2",
        "some_value_1": 2,
        "some_value_2": 3
    },
    {"key":"unique_key_3",
        "some_value_1": 2,
        "some_value_2": 1
    }
]

如果您无法控制 json，则可以将 from_json 列函数与 explode 一起使用。首先将 json 读取为单行单列文本，然后对其进行解析。

然后先用from_json解析文本：

json_schema = MapType(StringType(), StringType()) df.withColumn("json", from_json(col('text'), json_schema)) # expand into key-value column

然后，将新创建的对象的键分解成单独的行：

.select(explode(col('json'))) # make a row for each key in the json

最后，您可以执行相同的操作来解压缩值并将它们选择到单独的列中。这是一个将所有内容放在一起的小演示：

from pyspark.sql.types import *
from pyspark.sql.functions import *

text_schema = StructType([StructField('text', StringType(), True)])
json_schema = MapType(StringType(), StringType())

data = """{
    "unique_key_1": {
        "some_value_1": 1,
        "some_value_2": 2
    },
    "unique_key_2": {
        "some_value_1": 2,
        "some_value_2": 3
    },
    "unique_key_3": {
        "some_value_1": 2,
        "some_value_2": 1
    }
}
"""

df = (spark.createDataFrame([(data,)], schema=text_schema) # read dataframe
  .withColumn("json", from_json(col('text'), json_schema)) # expand into key-value column 
  .select(explode(col('json'))) # make a row for each key in the json
  .withColumn("value", from_json(col('value'), json_schema)) # now interpret the value for each key as json also
  .withColumn("some_value_1", col("value.some_value_1")) # unpack the object into separate rows
  .withColumn("some_value_2", col("value.some_value_2"))
  .drop('value')
     )

display(df)