如何将3个DataFrame合并转换为指定嵌套字典/JSON格式（Databricks）

Pandas 实现方式 #

1. 逐层构建嵌套结构 #

通过多层分组和自定义函数，从最内层的供应商数据开始向上构建：

import pandas as pd
import json

# 假设已完成三表合并，得到merged_df
# 定义函数：将单一部件的供应商数据转为字典列表
def build_supplier_list(group):
    return group[['supplier_name', 'cost']].to_dict('records')

# 定义函数：构建单个产品的完整嵌套结构
def build_product_struct(group):
    # 提取产品基础信息（取第一条即可，同一product_no的信息一致）
    product_info = group[['product_no', 'product_name', 'product_start_date', 'product_end_date', 'product_price']].iloc[0].to_dict()
    # 按部件分组，为每个部件添加供应商列表
    part_data = []
    for part_id, part_group in group.groupby('part_registration_no'):
        part_info = part_group[['part_registration_no', 'part_name', 'part_make', 'part_color']].iloc[0].to_dict()
        part_info['supplier_data'] = build_supplier_list(part_group)
        part_data.append(part_info)
    product_info['part_data'] = part_data
    return product_info

# 生成最终嵌套列表并转为JSON
final_result = merged_df.groupby('product_no').apply(build_product_struct).tolist()
json_output = json.dumps(final_result, default=str)  # default=str处理日期类型序列化

关键注意点 #

• 确保合并后的DataFrame没有重复的关联键数据，避免生成冗余嵌套条目
• 使用default=str处理日期、数值等非JSON原生类型，防止序列化报错

PySpark 实现方式（适配Databricks大数据场景） #

如果在Databricks中使用Spark，推荐用原生函数构建嵌套结构，性能更优：

from pyspark.sql import functions as F

# 假设已完成三表合并，得到spark_merged_df（Spark DataFrame）
# 第一步：为每个部件聚合供应商数据
part_with_suppliers = spark_merged_df.groupBy(
    'product_no', 'part_registration_no', 'part_name', 'part_make', 'part_color'
).agg(
    F.collect_list(F.struct('supplier_name', 'cost')).alias('supplier_data')
)

# 第二步：为每个产品聚合部件数据
final_spark_df = part_with_suppliers.groupBy(
    'product_no', 'product_name', 'product_start_date', 'product_end_date', 'product_price'
).agg(
    F.collect_list(F.struct(
        'part_registration_no', 'part_name', 'part_make', 'part_color', 'supplier_data'
    )).alias('part_data')
)

# 转换为JSON格式
# 方式1：获取JSON字符串数组
json_lines = final_spark_df.toJSON().collect()
# 方式2：合并为单个JSON数组
import json
full_json = json.dumps([json.loads(line) for line in json_lines])
# 方式3：直接写入Databricks存储
final_spark_df.write.mode('overwrite').json('/dbfs/path/to/save/json')

关键注意点 #

• 使用collect_list+struct组合是Spark构建嵌套结构的标准方式，适合大规模数据
• 若需处理空值，可提前用F.fillna或F.coalesce填充，避免生成空数组

常见问题排查 #

• 分组后数据重复：合并前检查各表的关联键是否唯一，比如Part表的part_registration_no是否唯一标识一个部件
• 日期序列化失败：Spark中可通过配置spark.sql.jsonGenerator.ignoreNullFields或手动将日期转为字符串
• 嵌套层级错误：确保分组顺序从内层（供应商）到外层（产品），避免层级颠倒

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