如何使用Spark和Scala将扁平化DataFrame转换为嵌套JSON

需求说明 #

给定如下结构的DataFrame：

需要转换为指定结构的嵌套JSON：

{
"shop_id":123,
"mng_number":"vj123",
"item_id":456,
"vj":{
"vj123-red": {
"dual_price_id":1,
"regular_price":1222222
},
"vj123-green": {
"dual_price_id":1,
"regular_price":238000
}
}
}

实现代码（Python + Pandas） #

import pandas as pd
import json

# 构建示例DataFrame
df = pd.DataFrame({
    'shop_id': [123, 123],
    'mng_number': ['vj123', 'vj123'],
    'item_id': [456, 456],
    'variant_id': ['vj123-red', 'vj123-green'],
    'dual_price_id': [1, 1],
    'regular_price': [1222222, 238000]
})

# 提取固定字段（假设每组数据的shop_id、mng_number、item_id一致）
base_data = df[['shop_id', 'mng_number', 'item_id']].iloc[0].to_dict()

# 构建variant对应的嵌套结构
variant_dict = {}
for _, row in df.iterrows():
    variant_dict[row['variant_id']] = {
        'dual_price_id': row['dual_price_id'],
        'regular_price': row['regular_price']
    }

# 合并到最终结果，这里的"vj"对应示例需求中的固定键名
base_data['vj'] = variant_dict

# 转换为JSON并格式化输出
result_json = json.dumps(base_data, indent=2)
print(result_json)

代码说明 #

• 先提取DataFrame中所有行共有的固定字段（shop_id、mng_number、item_id），直接取第一行数据转为字典。
• 遍历DataFrame每一行，将variant_id作为键，对应的价格相关字段组成字典作为值，构建嵌套的变体字典。
• 将变体字典合并到基础数据字典中，键名vj匹配示例需求；如果需要根据mng_number动态生成键名，可替换为base_data[row['mng_number'].split('')[0]] = variant_dict这类逻辑。
• 最后用json.dumps将字典格式化为带缩进的JSON字符串。

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