如何利用Pandas json_normalize扁平化嵌套JSON，转指定列后存为CSV

问题描述 #

给定包含嵌套数组的标准JSON数据：

[
    {
        "SKU":"TEST1",
        "name":"test name 1",
        "ItemSalesPrices":[
            {"SourceNumber": "OEM", "AssetNumber": "TEST1", "UnitPrice": 1600},
            {"SourceNumber": "RRP", "AssetNumber": "TEST1", "UnitPrice": 1500}
        ],
        "ItemDiscounts":[
            {"SourceNumber": "RETAIL", "AssetNumber": "AC", "LineDiscountPercentage": 30},
            {"SourceNumber": "LARGE ACC", "AssetNumber": "AC", "LineDiscountPercentage": 45}
        ]
    },
    {
        "SKU":"TEST2",
        "name":"test name 2",
        "ItemSalesPrices":[
            {"SourceNumber": "RRP", "AssetNumber": "TEST2", "UnitPrice": 1500}
        ],
        "ItemDiscounts":[
            {"SourceNumber": "RETAIL", "AssetNumber": "AC", "LineDiscountPercentage": 30}
        ]
    }
]

需要将其扁平化后导出为CSV，最终保留原始的2行数据，将数组中每个SourceNumber对应的值转为单独列，目标CSV结构如下：

原方法通过遍历每个条目和数组并校验SourceNumber的方式效率较低，以下是更合理的解决方案。

解决方案 #

方法1：使用Python pandas库（简洁高效） #

利用pandas的json_normalize快速解析嵌套JSON，再通过透视表转换为宽表，最后合并得到目标格式：

import pandas as pd

# 示例JSON数据（实际场景可从文件读取）
data = [
    {
        "SKU":"TEST1",
        "name":"test name 1",
        "ItemSalesPrices":[
            {"SourceNumber": "OEM", "AssetNumber": "TEST1", "UnitPrice": 1600},
            {"SourceNumber": "RRP", "AssetNumber": "TEST1", "UnitPrice": 1500}
        ],
        "ItemDiscounts":[
            {"SourceNumber": "RETAIL", "AssetNumber": "AC", "LineDiscountPercentage": 30},
            {"SourceNumber": "LARGE ACC", "AssetNumber": "AC", "LineDiscountPercentage": 45}
        ]
    },
    {
        "SKU":"TEST2",
        "name":"test name 2",
        "ItemSalesPrices":[
            {"SourceNumber": "RRP", "AssetNumber": "TEST2", "UnitPrice": 1500}
        ],
        "ItemDiscounts":[
            {"SourceNumber": "RETAIL", "AssetNumber": "AC", "LineDiscountPercentage": 30}
        ]
    }
]

# 解析价格数组并转为宽表
prices_df = pd.json_normalize(data, record_path='ItemSalesPrices', meta=['SKU'])
prices_wide = prices_df.pivot(index='SKU', columns='SourceNumber', values='UnitPrice').reset_index()
prices_wide.columns = ['SKU'] + [f'ItemSalesPrices_{col}' for col in prices_wide.columns[1:]]

# 解析折扣数组并转为宽表
discounts_df = pd.json_normalize(data, record_path='ItemDiscounts', meta=['SKU'])
discounts_wide = discounts_df.pivot(index='SKU', columns='SourceNumber', values='LineDiscountPercentage').reset_index()
discounts_wide.columns = ['SKU'] + [f'ItemDiscounts_{col}' for col in discounts_wide.columns[1:]]

# 合并基础信息与转换后的宽表
base_df = pd.DataFrame(data)[['SKU', 'name']].rename(columns={'name': 'Name'})
final_df = base_df.merge(prices_wide, on='SKU').merge(discounts_wide, on='SKU')

# 调整列顺序并填充空值
target_columns = [
    'SKU', 'Name', 'ItemSalesPrices_OEM', 'ItemSalesPrices_RRP',
    'ItemDiscounts_RETAIL', 'ItemDiscounts_LARGE ACC'
]
final_df = final_df.reindex(columns=target_columns).fillna('')

# 导出为CSV文件
final_df.to_csv('flattened_data.csv', index=False)

方法2：纯Python实现（无需第三方库） #

如果无法使用第三方库，可通过预定义目标列，遍历每个条目并映射对应值：

import csv

# 示例JSON数据
data = [
    {
        "SKU":"TEST1",
        "name":"test name 1",
        "ItemSalesPrices":[
            {"SourceNumber": "OEM", "AssetNumber": "TEST1", "UnitPrice": 1600},
            {"SourceNumber": "RRP", "AssetNumber": "TEST1", "UnitPrice": 1500}
        ],
        "ItemDiscounts":[
            {"SourceNumber": "RETAIL", "AssetNumber": "AC", "LineDiscountPercentage": 30},
            {"SourceNumber": "LARGE ACC", "AssetNumber": "AC", "LineDiscountPercentage": 45}
        ]
    },
    {
        "SKU":"TEST2",
        "name":"test name 2",
        "ItemSalesPrices":[
            {"SourceNumber": "RRP", "AssetNumber": "TEST2", "UnitPrice": 1500}
        ],
        "ItemDiscounts":[
            {"SourceNumber": "RETAIL", "AssetNumber": "AC", "LineDiscountPercentage": 30}
        ]
    }
]

# 定义目标CSV列
target_columns = [
    'SKU', 'Name', 'ItemSalesPrices_OEM', 'ItemSalesPrices_RRP',
    'ItemDiscounts_RETAIL', 'ItemDiscounts_LARGE ACC'
]

flattened_rows = []
for item in data:
    # 初始化行字典，默认空值
    row = {col: '' for col in target_columns}
    # 填充基础字段
    row['SKU'] = item['SKU']
    row['Name'] = item['name']
    
    # 映射价格数据到对应列
    for price in item['ItemSalesPrices']:
        col_key = f'ItemSalesPrices_{price["SourceNumber"]}'
        if col_key in row:
            row[col_key] = price['UnitPrice']
    
    # 映射折扣数据到对应列
    for discount in item['ItemDiscounts']:
        col_key = f'ItemDiscounts_{discount["SourceNumber"]}'
        if col_key in row:
            row[col_key] = discount['LineDiscountPercentage']
    
    flattened_rows.append(row)

# 写入CSV文件
with open('flattened_data.csv', 'w', newline='', encoding='utf-8') as csv_file:
    writer = csv.DictWriter(csv_file, fieldnames=target_columns)
    writer.writeheader()
    writer.writerows(flattened_rows)

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