如何将PySpark DataFrame转换为多层嵌套JSON

完全可以实现，不需要分组操作，直接用PySpark的struct函数就能逐层构造嵌套结构，以下是具体实现步骤：

实现步骤 #

1. 导入依赖：需要用到struct函数来构造嵌套列
2. 逐层嵌套结构：
   • 先把docdetailfield1和docdetailfield2封装到docDetails子结构中
   • 再把docname和docDetails封装到docsection结构中
   • 最后保留transactionid和嵌套后的docsection列
3. 输出JSON：可以写出到文件，或者直接获取每行的JSON字符串

完整代码示例 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import struct, col

# 初始化SparkSession（如果已有可跳过）
spark = SparkSession.builder.appName("NestedJSONConversion").getOrCreate()

# 替换为你从SQL读取的DataFrame
# 示例DataFrame
sample_data = [
    (1234, "name1", "name1detail1", "name1detail2"),
    (1235, "name2", "name2detail1", "name2detail2")
]
sample_cols = ["transactionid", "docname", "docdetailfield1", "docdetailfield2"]
df = spark.createDataFrame(sample_data, sample_cols)

# 构造嵌套结构
nested_df = df.select(
    col("transactionid"),
    struct(
        col("docname"),
        struct(
            col("docdetailfield1"),
            col("docdetailfield2")
        ).alias("docDetails")
    ).alias("docsection")
)

# 验证结构（可选）
nested_df.printSchema()

# 输出为JSON文件，替换为你的输出路径
nested_df.write.mode("overwrite").json("/your/output/path")

# 或者打印每行的JSON字符串
for json_row in nested_df.toJSON().collect():
    print(json_row)

结构验证 #

执行printSchema()后，输出的结构如下，完全匹配你的嵌套需求：

root
 |-- transactionid: long (nullable = true)
 |-- docsection: struct (nullable = false)
 |    |-- docname: string (nullable = true)
 |    |-- docDetails: struct (nullable = false)
 |    |    |-- docdetailfield1: string (nullable = true)
 |    |    |-- docdetailfield2: string (nullable = true)

说明 #

你之前尝试分组得到错误结构，是因为分组适用于将多行数据聚合到一个嵌套数组或结构中，而你的场景中每个transactionid对应唯一的一行数据，直接用struct嵌套列即可完成转换，不需要分组操作。

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