Pandas嵌套JSON归一化：添加product_seq与父产品序列ID

问题描述 #

需要对包含嵌套JSON对象的DataFrame进行归一化展开，同时为每个产品添加两个字段：

• product_seq：全局唯一的产品自增序列ID
• parent_product_seq：对应父产品的序列ID（根产品的父ID为0）

示例输入JSON #

{"subscription":[{"subscription_boo":true,"subscription_ref_prefix":"S_","product":[{"no_of_products":1,"product_id":1,"product":[{"no_of_products":1,"product_id":1.1,"product":[{"no_of_products":1,"product_id":1.11}]},{"no_of_products":1,"product_id":1.2}]}]},{"subscription_boo":true,"subscription_ref_prefix":"B_","product":[{"no_of_products":1,"product_id":2,"product":[{"no_of_products":1,"product_id":2.1,"product":[{"no_of_products":1,"product_id":2.11,"product":[{"no_of_products":1,"product_id":2.11}]},{"no_of_products":1,"product_id":2.11}]},{"no_of_products":1,"product_id":2.2}]}]}]}

期望输出 #

解决方案代码 #

import pandas as pd

def traverse_products(products, subscription_data, parent_seq, seq_counter):
    rows = []
    for prod in products:
        # 构造当前产品行数据
        current_row = {
            "subscription_boo": subscription_data["subscription_boo"],
            "subscription_ref_prefix": subscription_data["subscription_ref_prefix"],
            "no_of_products": prod["no_of_products"],
            "product_id": prod["product_id"],
            "product_seq": seq_counter[0],
            "parent_product_seq": parent_seq
        }
        rows.append(current_row)
        # 计数器自增
        seq_counter[0] += 1
        # 递归处理子产品
        if "product" in prod and prod["product"]:
            rows.extend(traverse_products(prod["product"], subscription_data, seq_counter[0]-1, seq_counter))
    return rows

# 加载示例JSON数据
input_json = {"subscription":[{"subscription_boo":True,"subscription_ref_prefix":"S_","product":[{"no_of_products":1,"product_id":1,"product":[{"no_of_products":1,"product_id":1.1,"product":[{"no_of_products":1,"product_id":1.11}]},{"no_of_products":1,"product_id":1.2}]}]},{"subscription_boo":True,"subscription_ref_prefix":"B_","product":[{"no_of_products":1,"product_id":2,"product":[{"no_of_products":1,"product_id":2.1,"product":[{"no_of_products":1,"product_id":2.11,"product":[{"no_of_products":1,"product_id":2.11}]},{"no_of_products":1,"product_id":2.11}]},{"no_of_products":1,"product_id":2.2}]}]}]}

# 初始化结果列表和序列计数器（用列表实现可变对象传递）
result_rows = []
seq_counter = [1]

# 遍历每个订阅项
for sub in input_json["subscription"]:
    if sub.get("product"):
        result_rows.extend(traverse_products(sub["product"], sub, 0, seq_counter))

# 转换为DataFrame并格式化
df = pd.DataFrame(result_rows)
# 格式化product_id为两位小数
df["product_id"] = df["product_id"].apply(lambda x: f"{x:.2f}").astype(float)
# 调整列顺序匹配期望输出
df = df[["subscription_boo", "subscription_ref_prefix", "no_of_products", "product_id", "product_seq", "parent_product_seq"]]

print(df.to_string(index=False))

代码说明 #

1. 递归遍历函数：traverse_products负责递归处理嵌套的产品结构，每次处理一个产品时生成对应的行数据，传递父产品的序列ID，并维护全局自增的序列计数器。
2. 计数器处理：使用列表seq_counter作为可变对象传递，确保递归过程中计数器能持续自增。
3. 数据格式化：最后将收集的行数据转换为DataFrame，对product_id进行两位小数格式化，并调整列顺序匹配期望输出。

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