PySpark处理嵌套JSON数组教程

怎么入门文章编程？需要学习哪些知识点？这是新手们刚接触编程时常见的问题；下面golang学习网就来给大家整理分享一些知识点，希望能够给初学者一些帮助。本篇文章就来介绍《PySpark转嵌套JSON数组教程》，涉及到，有需要的可以收藏一下

引言

在数据处理和集成场景中，将关系型或扁平化的数据结构转换为具有层次感的JSON格式是一项常见的需求。特别是在处理订单明细、商品列表等具有“一主多从”关系的数据时，需要将多个关联的行聚合成一个嵌套的JSON数组。PySpark作为大数据处理的强大工具，提供了丰富的API来高效完成这类复杂的数据转换。本教程将以一个具体的例子，演示如何将一个包含订单及其多个商品项的PySpark DataFrame，转换为一个嵌套的JSON数组结构。

原始数据结构与目标JSON格式

假设我们有一个PySpark DataFrame，其结构如下所示，其中每个订单项（由itemSeqNo区分）的属性（Date, Amount, description）以行式存储：

原始DataFrame示例：

我们的目标是将其转换为以下嵌套的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()

注意事项与最佳实践

1. 数据类型匹配： 在使用F.struct()和F.collect_list()时，确保列的数据类型符合预期。如果需要，可以使用cast()函数进行类型转换。
2. 列名一致性： 确保pivot操作后生成的列名与目标JSON结构中的键名一致。
3. 性能优化： 对于大规模数据集，pivot操作可能会消耗大量内存和计算资源。如果field列的唯一值非常多，pivot可能不是最佳选择。在这种情况下，可以考虑其他方法，例如使用map类型或自定义UDF（用户定义函数），但通常内置函数性能更优。
4. 空值处理： 在聚合和结构化过程中，PySpark会根据默认行为处理空值。如果需要特定的空值处理逻辑（例如，在JSON中省略空字段），可能需要在生成Struct之前进行过滤或使用when().otherwise()。
5. Schema定义： 在创建Struct时，PySpark会自动推断Schema。如果需要更严格的Schema控制或处理复杂类型，可以显式定义StructType。
6. 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数据转换技巧，将极大地提升您在大数据平台上的数据处理效率和灵活性。

今天关于《PySpark处理嵌套JSON数组教程》的内容就介绍到这里了，是不是学起来一目了然！想要了解更多关于的内容请关注golang学习网公众号！