如何使用PySpark将特定结构的DataFrame转换为目标格式JSON文件

实现思路 #

要把给定的长格式DataFrame转换成嵌套键值对的JSON，核心是先按Key分组聚合，将每组的desc和value映射为子对象，再将所有Key与对应子对象整合成全局结构。

具体代码实现 #

1. 分组构造子属性Map
   先按Key分组，把每组的desc和value收集成数组，再转成Map类型：

from pyspark.sql import functions as F

# 替换成你的DataFrame名称
grouped_df = df.groupBy("Key").agg(
    F.map_from_arrays(
        F.collect_list("desc"),
        F.collect_list("value")
    ).alias("attrs")
)

执行后得到的DataFrame结构：

2. 合并为全局键值对并转JSON
   将所有Key和对应的attrs合并成一个大Map，再转换成JSON字符串：

# 聚合所有行生成全局Map
final_map_df = grouped_df.agg(
    F.map_from_arrays(
        F.collect_list(F.col("Key").cast("string")),  # 确保Key是字符串类型
        F.collect_list("attrs")
    ).alias("final_result")
)

# 提取JSON字符串
target_json = final_map_df.select(F.to_json("final_result")).first()[0]

3. 写入JSON文件
   如果需要将结果保存到文件，可创建单行DataFrame后写入：

from pyspark.sql.types import StringType

# 生成仅含目标JSON的DataFrame
json_output_df = spark.createDataFrame([(target_json,)], StringType())

# 写入指定路径，mode可根据需求设为overwrite/append等
json_output_df.write.mode("overwrite").text("/your/output/path")

注意点 #

• 必须将Key转为字符串类型，否则生成的JSON中键会是数字而非带引号的字符串，不符合需求格式。
• 若处理超大数据集，全量聚合到单个分区可能导致性能问题，此时可考虑先按Key输出小文件，再用Python的json库合并结果，小数据量场景直接用上述方法即可。

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