引言
在数据处理和集成场景中，将关系型或扁平化的数据结构转换为具有层次感的json格式是一项常见的需求。特别是在处理订单明细、商品列表等具有“一主多从”关系的数据时，需要将多个关联的行聚合成一个嵌套的json数组。pyspark作为大数据处理的强大工具，提供了丰富的api来高效完成这类复杂的数据转换。本教程将以一个具体的例子，演示如何将一个包含订单及其多个商品项的pyspark dataframe，转换为一个嵌套的json数组结构。
原始数据结构与目标JSON格式
假设我们有一个PySpark DataFrame，其结构如下所示，其中每个订单项（由itemSeqNo区分）的属性（Date, Amount, description）以行式存储：
原始DataFrame示例：
| OrderID | field | fieldValue | itemSeqNo |
|---|---|---|---|
| 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 |
我们的目标是将其转换为以下嵌套的JSON结构：
目标JSON结构：
{
"orderDetails": {
"orderID": "123"
},
"itemizationDetails": [
{
"Date": "01-01-23",
"Amount": "10.00",
"description": "Pencil"
},
{
"Date": "01-02-23",
"Amount": "11.00",
"description": "Pen"
}
]
}可以看到，itemizationDetails是一个数组，其中每个元素代表一个订单项，其属性（Date, Amount, description）被聚合到单个对象中。
PySpark 转换步骤详解
我们将分步实现上述转换。首先，确保您已经启动了SparkSession并导入了必要的PySpark函数。
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType
# 初始化SparkSession
spark = SparkSession.builder.appName("NestedJsonConversion").getOrCreate()
# 创建示例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")
]
schema = StructType([
StructField("OrderID", StringType(), True),
StructField("field", StringType(), True),
StructField("fieldValue", StringType(), True),
StructField("itemSeqNo", StringType(), True)
])
df = spark.createDataFrame(data, schema)
df.show()步骤1：重塑DataFrame (Pivot操作)
首先，我们需要将每个订单项的属性（如Date, Amount, description）从行转换为列。这可以通过groupBy结合pivot操作实现。pivot需要一个聚合函数，这里我们使用F.first()来获取fieldValue，因为每个field在OrderID和itemSeqNo的组合下应该只有一个fieldValue。
df_pivoted = df.groupBy('OrderID', 'itemSeqNo').pivot('field').agg(F.first('fieldValue'))
df_pivoted.show()
# 预期输出：
# +-------+---------+------+---------+-----------+
# |OrderID|itemSeqNo|Amount| Date|description|
# +-------+---------+------+---------+-----------+
# | 123| 1| 10.00| 01-01-23| Pencil|
# | 123| 2| 11.00|01-02-23 | Pen |
# +-------+---------+------+---------+-----------+这一步将原本扁平化的数据结构转换成了每个订单项一行，所有相关属性作为列的宽表格式，为后续的结构化操作打下基础。
步骤2：将订单项详情打包为Struct类型
接下来，我们将每个订单项的各个属性（Amount, Date, description）打包成一个名为itemizationDetails的Struct（结构体）类型列。这使得每个订单项的完整信息可以作为一个独立的嵌套对象处理。
df_item_struct = df_pivoted.withColumn(
'itemizationDetails',
F.struct(F.col('Amount'), F.col('Date'), F.col('description'))
)
df_item_struct.show(truncate=False)
# 预期输出：
# +-------+---------+------+---------+-----------+-------------------------+
# |OrderID|itemSeqNo|Amount|Date |description|itemizationDetails |
# +-------+---------+------+---------+-----------+-------------------------+
# |123 |1 |10.00 |01-01-23 |Pencil |{10.00, 01-01-23, Pencil}|
# |123 |2 |11.00 |01-02-23 |Pen |{11.00, 01-02-23 , Pen } |
# +-------+---------+------+---------+-----------+-------------------------+通过F.struct()函数，我们有效地创建了一个嵌套的数据结构，其中包含了单个订单项的所有相关信息。
步骤3：按订单ID收集订单项列表
现在，我们需要将同一个OrderID下的所有itemizationDetails Struct收集到一个列表中，形成JSON中的itemizationDetails数组。这通过再次groupBy OrderID并使用F.collect_list()聚合函数实现。
df_collected_list = df_item_struct.groupBy('OrderID').agg(
F.collect_list('itemizationDetails').alias('itemizationDetails')
)
df_collected_list.show(truncate=False)
# 预期输出：
# +-------+-----------------------------------------------------+
# |OrderID|itemizationDetails |
# +-------+-----------------------------------------------------+
# |123 |[{10.00, 01-01-23, Pencil}, {11.00, 01-02-23 , Pen }]|
# +-------+-----------------------------------------------------+F.collect_list()是创建JSON数组的关键，它将所有聚合的Struct对象收集成一个ArrayType列。
步骤4：将订单ID打包为Struct类型
为了符合目标JSON中orderDetails的嵌套结构，我们需要将OrderID也打包成一个Struct类型。
df_final_struct = df_collected_list.withColumn('orderDetails', F.struct(F.col('OrderID')))
df_final_struct.show(truncate=False)
# 预期输出：
# +-------+-----------------------------------------------------+------------+
# |OrderID|itemizationDetails |orderDetails|
# +-------+-----------------------------------------------------+------------+
# |123 |[{10.00, 01-01-23, Pencil}, {11.00, 01-02-23 , Pen }]|{123} |
# +-------+-----------------------------------------------------+------------+这一步创建了顶层orderDetails对象。
步骤5：导出DataFrame到JSON
最后一步是将处理好的DataFrame导出为JSON格式。我们只需要选择orderDetails和itemizationDetails这两列，然后使用toJSON().collect()方法即可。
result_json_rdd = df_final_struct.select('orderDetails', 'itemizationDetails').toJSON()
result_list = result_json_rdd.collect()
for json_str in result_list:
print(json_str)
# 预期输出（可能格式化略有不同，但内容一致）：
# {"orderDetails":{"OrderID":"123"},"itemizationDetails":[{"Amount":"10.00","Date":"01-01-23","description":"Pencil"},{"Amount":"11.00","Date":"01-02-23 ","description":"Pen "}]}toJSON()方法会将DataFrame的每一行转换为一个JSON字符串。collect()则将这些字符串收集到一个Python列表中。
完整代码示例
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType
# 初始化SparkSession
spark = SparkSession.builder.appName("NestedJsonConversion").getOrCreate()
# 1. 创建示例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")
]
schema = StructType([
StructField("OrderID", StringType(), True),
StructField("field", StringType(), True),
StructField("fieldValue", StringType(), True),
StructField("itemSeqNo", StringType(), True)
])
df = spark.createDataFrame(data, schema)
print("--- 原始 DataFrame ---")
df.show()
# 2. 重塑DataFrame：将 field 列的值转换为列名
df_pivoted = df.groupBy('OrderID', 'itemSeqNo').pivot('field').agg(F.first('fieldValue'))
print("--- Pivot 后的 DataFrame ---")
df_pivoted.show()
# 3. 将订单项详情打包为Struct类型
df_item_struct = df_pivoted.withColumn(
'itemizationDetails',
F.struct(F.col('Amount'), F.col('Date'), F.col('description'))
)
print("--- itemizationDetails Struct 创建后的 DataFrame ---")
df_item_struct.show(truncate=False)
# 4. 按订单ID收集订单项列表
df_collected_list = df_item_struct.groupBy('OrderID').agg(
F.collect_list('itemizationDetails').alias('itemizationDetails')
)
print("--- 收集 itemizationDetails 列表后的 DataFrame ---")
df_collected_list.show(truncate=False)
# 5. 将订单ID打包为Struct类型
df_final_struct = df_collected_list.withColumn('orderDetails', F.struct(F.col('OrderID')))
print("--- orderDetails Struct 创建后的 DataFrame ---")
df_final_struct.show(truncate=False)
# 6. 导出DataFrame到JSON
result_json_rdd = df_final_struct.select('orderDetails', 'itemizationDetails').toJSON()
result_list = result_json_rdd.collect()
print("\n--- 最终 JSON 输出 ---")
for json_str in result_list:
import json
# 为了更好的可读性，这里对JSON字符串进行美化打印
print(json.dumps(json.loads(json_str), indent=3, ensure_ascii=False))
# 停止SparkSession
spark.stop()注意事项与最佳实践
- 数据类型匹配： 在使用F.struct()和F.collect_list()时，确保列的数据类型符合预期。如果需要，可以使用cast()函数进行类型转换。
- 列名一致性： 确保pivot操作后生成的列名与目标JSON结构中的键名一致。
- 性能优化： 对于大规模数据集，pivot操作可能会消耗大量内存和计算资源。如果field列的唯一值非常多，pivot可能不是最佳选择。在这种情况下，可以考虑其他方法，例如使用map类型或自定义UDF（用户定义函数），但通常内置函数性能更优。
- 空值处理： 在聚合和结构化过程中，PySpark会根据默认行为处理空值。如果需要特定的空值处理逻辑（例如，在JSON中省略空字段），可能需要在生成Struct之前进行过滤或使用when().otherwise()。
- Schema定义： 在创建Struct时，PySpark会自动推断Schema。如果需要更严格的Schema控制或处理复杂类型，可以显式定义StructType。
- toJSON()与write.json()： toJSON().collect()适用于将结果收集到驱动程序内存中进行进一步处理或打印。对于将大量数据直接写入文件系统（如HDFS、S3）的场景，推荐使用df.write.json("output_path")，它能以分布式方式写入，且每行一个JSON对象。如果需要一个包含所有JSON对象的单个文件，可能需要先coalesce(1)再写入。
总结
通过本教程，我们学习了如何利用PySpark的pivot、struct和collect_list等核心函数，将一个扁平化的DataFrame逐步转换为具有复杂嵌套结构和数组的JSON格式。这种转换能力在处理来自关系型数据库的数据，并将其适配到API接口、文档型数据库或消息队列等需要层次化数据的场景中至关重要。掌握这些PySpark数据转换技巧，将极大地提升您在大数据平台上的数据处理效率和灵活性。
