如何用PySpark解析并扁平化嵌套JSON数据？求解决方案

我正尝试使用PySpark解析并扁平化嵌套JSON数据，但当前代码未能成功，现寻求可行方案。以下是待解析的JSON数据、已尝试的代码片段以及期望的扁平化输出结果：

待解析JSON数据 #

jsonData ="""{
    "data": {
        "unique_id1": {
            "random_code1": {
                "name": "some_name",
                "status": "value1"
            },
            "random_code2": {
                "name": "some_name",
                "status": "value2"
            }
        },
        "unique_id2": {
            "random_code3": {
                "name": "some_name",
                "status": "value2"
            },
            "random_code4": {
                "name": "some_name",
                "status": "value2"
            }
        }
    }
}"""

已尝试的代码片段 #

df = spark.read.option("multiLine", "true").json(spark.sparkContext.parallelize([jsonData]))

data_schema = df.schema["data"].dataType.simpleString()
data_schema  = re.sub(r"([\w\-]+)(?=:struct<name)", "_RandomCode", data_schema)
data_schema  = re.sub(r"([\w\-]+)(?=:struct<_RandomCode)", "_Ids", data_schema)
data_schema = re.sub(r"(?<=,|<)([^,<]+)(?=:)", r"`\1`", data_schema)

期望输出 #

_Ids              _RandomCode              name
unique_id1         random_code1            some_name
unique_id1         random_code2            some_name
unique_id2         random_code3            some_name
unique_id2         random_code4            some_name

可行解析方案 #

由于JSON中的unique_id*和random_code*都是动态生成的键，无法通过固定字段名直接解析，需要用PySpark内置函数动态展开嵌套结构，步骤如下：

1. 展开第一层动态键（unique_id）：用map_keys获取data字段下的所有键，通过explode将每个键值对拆分为行，保留键名作为_Ids字段。
2. 展开第二层动态键（random_code）：对第一步得到的value字段，再次用map_keys和explode拆分，保留键名作为_RandomCode字段。
3. 提取嵌套字段：从最终的value中提取name字段，完成扁平化。

完整代码示例 #

from pyspark.sql import functions as F

# 读取JSON数据
df = spark.read.option("multiLine", "true").json(spark.sparkContext.parallelize([jsonData]))

# 第一步：展开unique_id层
df_step1 = df.select(
    F.explode(F.map_keys(df.data)).alias("_Ids"),
    F.element_at(F.map_values(df.data), F.expr("position(map_keys(data), _Ids)")).alias("random_code_map")
)

# 第二步：展开random_code层，并提取name字段
final_df = df_step1.select(
    "_Ids",
    F.explode(F.map_keys(df_step1.random_code_map)).alias("_RandomCode"),
    F.element_at(F.map_values(df_step1.random_code_map), F.expr("position(map_keys(random_code_map), _RandomCode)")).alias("details")
).select(
    "_Ids",
    "_RandomCode",
    F.col("details.name").alias("name")
)

# 查看结果
final_df.show()

执行结果 #

+----------+------------+----------+
|      _Ids|_RandomCode|      name|
+----------+------------+----------+
|unique_id1|random_code1|some_name|
|unique_id1|random_code2|some_name|
|unique_id2|random_code3|some_name|
|unique_id2|random_code4|some_name|
+----------+------------+----------+

方案说明 #

• map_keys：获取Map类型字段的所有键，适配动态键场景。
• explode：将Map的元素拆分为多行，实现层级扁平化。
• element_at：通过键的位置匹配对应的value，避免explode(map_values)导致的行重复问题。

内容的提问来源于stack exchange，提问作者Idleguys