PySpark多级JSON转DataFrame：如何将嵌套字段展开为独立列

我来帮你搞定这个问题！你现在遇到的问题主要是两个点：一是JSON读取方式导致嵌套结构解析异常，二是没对嵌套的Address和数组类型的Pets做针对性处理。下面一步步来解决：

1. 先修正JSON读取方式 #

你之前用sc.wholeTextFiles的方式其实没必要，PySpark的spark.read.json可以直接读取文件路径，它能自动识别嵌套的JSON结构，不会把Address这种对象类型误解析成数组：

path_to_input = "/FileStore/tables/sample_json_file2-6c20f.json"
df = spark.read.json(path_to_input)
# 先看一下正确的Schema
df.printSchema()

这时候输出的Schema会正确识别Address是结构体，Pets是数组：

root
 |-- Address: struct (nullable = true)
 |    |-- Permanent address: string (nullable = true)
 |    |-- current Address: string (nullable = true)
 |-- Boolean: boolean (nullable = true)
 |-- Mobile: long (nullable = true)
 |-- Name: string (nullable = true)
 |-- Pets: array (nullable = true)
 |    |-- element: string (containsNull = true)

2. 展开嵌套的Address字段 #

要把Address里的子字段拆成独立列，有两种常用方法：

方法一：手动提取并重命名 #

适合字段不多的情况，直接指定要提取的子字段，然后给列名加上前缀：

from pyspark.sql.functions import col

df_expanded = df.select(
    col("Name"),
    col("Mobile"),
    col("Boolean"),
    col("Address.Permanent address").alias("Address_Permanent_address"),
    col("Address.current Address").alias("Address_current_Address"),
    col("Pets")
)
df_expanded.show()

输出会变成：

+----+--------+-------+-------------------------+-----------------------+----------+
|Name|  Mobile|Boolean|Address_Permanent_address|Address_current_Address|      Pets|
+----+--------+-------+-------------------------+-----------------------+----------+
|Test|12345678|   true|                      USA|                     AU|[Dog, cat]|
+----+--------+-------+-------------------------+-----------------------+----------+

方法二：批量展开结构体 #

如果嵌套字段很多，用Address.*可以一次性展开所有子字段，之后再重命名即可：

df_expanded = df.select(
    "*",
    col("Address.*")
).drop("Address") \
 .withColumnRenamed("Permanent address", "Address_Permanent_address") \
 .withColumnRenamed("current Address", "Address_current_Address")
df_expanded.show()

这个方法和上面的效果完全一样，只是更高效。

3. 把Pets数组转成非数组形式 #

你说不想让Pets以数组展示，这里分两种常见需求：

需求A：把数组拆成多行（每个宠物一行） #

这是最常用的处理方式，用explode函数把数组的每个元素拆成单独的行，同时保留其他字段：

from pyspark.sql.functions import explode

df_final = df_expanded.select(
    "*",
    explode(col("Pets")).alias("Pet")
).drop("Pets")
df_final.show()

最终输出：

+----+--------+-------+-------------------------+-----------------------+----+
|Name|  Mobile|Boolean|Address_Permanent_address|Address_current_Address| Pet|
+----+--------+-------+-------------------------+-----------------------+----+
|Test|12345678|   true|                      USA|                     AU| Dog|
|Test|12345678|   true|                      USA|                     AU| cat|
+----+--------+-------+-------------------------+-----------------------+----+

需求B：把数组元素转成独立列（比如Pet_1、Pet_2） #

如果你的数组长度是固定的，也可以直接按索引提取元素，转成单独的列：

df_final = df_expanded.select(
    "*",
    col("Pets")[0].alias("Pet_1"),
    col("Pets")[1].alias("Pet_2")
).drop("Pets")
df_final.show()

输出会是：

+----+--------+-------+-------------------------+-----------------------+-----+-----+
|Name|  Mobile|Boolean|Address_Permanent_address|Address_current_Address|Pet_1|Pet_2|
+----+--------+-------+-------------------------+-----------------------+-----+-----+
|Test|12345678|   true|                      USA|                     AU|  Dog|  cat|
+----+--------+-------+-------------------------+-----------------------+-----+-----+

为啥原来的代码会出问题？ #

你用sc.wholeTextFiles(path_to_input).values()读取的是整个文件的字符串内容，这种方式更适合处理每行一个JSON对象的文件（比如JSON Lines格式）。但你的文件是单个完整的JSON对象，直接用spark.read.json(path_to_input)就能正确解析嵌套结构，不会把Address这种结构体误当成数组来处理。

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