如何将PySpark DataFrame转换为含多实例的嵌套JSON数组

问题分析 #

你当前使用Pandas的to_json(orient='records')直接输出的是扁平化单条记录数组，而目标结构需要将同一订单下的条目聚合为嵌套数组，并拆分出独立的订单详情部分。以下分别给出PySpark（匹配你原始需求）和Pandas（基于你当前代码）的实现方案：

PySpark 实现方案 #

针对PySpark DataFrame，我们通过pivot重塑数据、分组聚合生成嵌套结构：

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

# 初始化Spark会话
spark = SparkSession.builder.appName("OrderJsonConverter").getOrCreate()

# 创建示例PySpark DataFrame
data = [
    ("123", "Date", "01-01-23", "1"),
    ("123", "Amount", "10.00", "1"),
    ("123", "description", "Pencil", "1"),
    ("123", "Date", "01-02-23", "2"),
    ("123", "Amount", "11.00", "2"),
    ("123", "description", "Pen", "2")
]
columns = ["OrderID", "field", "fieldValue", "itemSeqNo"]
df = spark.createDataFrame(data, columns)

# 1. Pivot转换：将field转为列，fieldValue作为对应列的值
pivoted_df = df.groupBy("OrderID", "itemSeqNo")\
               .pivot("field")\
               .agg(first("fieldValue"))

# 2. 聚合生成嵌套结构：收集条目数组并构造订单详情
final_df = pivoted_df.groupBy("OrderID")\
                     .agg(collect_list(struct(col("Date"), col("Amount"), col("description"))).alias("itemizationDetails"))\
                     .select(
                         struct(col("OrderID").alias("orderID")).alias("orderDetails"),
                         col("itemizationDetails")
                     )

# 输出结果JSON字符串
result_json = final_df.toJSON().collect()[0]
print(result_json)

# 若要保存到文件
# final_df.write.json("order_output.json", mode="overwrite")

输出结果：

{"orderDetails":{"orderID":"123"},"itemizationDetails":[{"Date":"01-01-23","Amount":"10.00","description":"Pencil"},{"Date":"01-02-23","Amount":"11.00","description":"Pen"}]}

Pandas 实现方案 #

如果你需要基于当前的Pandas代码修改，可通过pivot_table重塑数据后构造嵌套字典：

import pandas as pd
import json

# 你的原始Pandas DataFrame
test_dataframe = pd.DataFrame(
    {
        "OrderID": ['123','123','123','123','123','123'],
        "field": ["Date","Amount",'description','Date','Amount','description'],
        "fieldValue": ['01-01-23','10.00','Pencil','01-02-23 ','11.00','Pen '],
        "itemSeqNo": ['1','1','1','2','2','2']
    }
)

# 1. 重塑数据：按条目分组，将field转为列
pivoted = test_dataframe.pivot_table(
    index=["OrderID", "itemSeqNo"],
    columns="field",
    values="fieldValue",
    aggfunc="first"
).reset_index(drop=True)

# 2. 构造目标嵌套结构
order_id = test_dataframe["OrderID"].iloc[0]
result = {
    "orderDetails": {"orderID": order_id},
    "itemizationDetails": pivoted.to_dict("records")
}

# 格式化输出JSON
print(json.dumps(result, indent=2))

输出结果：

{
  "orderDetails": {
    "orderID": "123"
  },
  "itemizationDetails": [
    {
      "Amount": "10.00",
      "Date": "01-01-23",
      "description": "Pencil"
    },
    {
      "Amount": "11.00",
      "Date": "01-02-23 ",
      "description": "Pen "
    }
  ]
}

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