如何使用Pandas动态归一化深度嵌套JSON/字典？

先修正硬编码的错误写法 #

你当前代码的meta参数使用逻辑错误：record_path=["test","broodje"]指向的是test数组元素下的broodje数组，此时landje是test数组元素的字段，而coupon、days_due等是broodje元素本身的字段，不需要放到meta里。另外aap是broodje里的嵌套数组，需要额外处理。

正确的硬编码处理步骤如下：

import pandas as pd
from pandas import json_normalize

data = {
   "test": [
       {
           "landje": "ES",
           "broodje": [
               {"coupon": "abcde", "aap": [{"type": "dirk", "dirk_code": "1234"}], "days_due": ["1"]},
               {"coupon": "zxcv", "aap": [{"type": "dirk", "dirk_code": "2345"}], "days_due": ["2"]}
           ]
       },
       {
           "landje": "FR",
           "broodje": [
               {"coupon": "ergd", "aap": [{"type": "dirk", "dirk_code": "4567"}], "days_due": ["3"]},
               {"coupon": "tygh", "aap": [{"type": "dirk", "dirk_code": "7789"}], "days_due": ["4"]}
           ]
       }
   ]
}

# 第一步：处理test -> broodje的层级，提取landje作为元数据
df_level1 = json_normalize(data["test"], record_path="broodje", meta="landje")

# 第二步：处理broodje下的aap嵌套数组
df_aap = json_normalize(df_level1["aap"].explode().tolist())

# 第三步：处理days_due单元素数组，提取值
df_days = df_level1["days_due"].str[0].rename("days_due")

# 合并所有结果
df_final = pd.concat([df_level1.drop(columns=["aap", "days_due"]), df_aap, df_days], axis=1)

# 调整列顺序（可选）
df_final = df_final[["landje", "coupon", "type", "dirk_code", "days_due"]]
print(df_final)

运行后会得到符合预期的结果：

landje coupon   type dirk_code days_due
0     ES  abcde   dirk      1234        1
1     ES   zxcv   dirk      2345        2
2     FR   ergd   dirk      4567        3
3     FR   tygh   dirk      7789        4

实现动态配置方案 #

通过定义配置字典指定各层级的处理规则，无需硬编码路径和字段：

1. 定义配置结构 #

配置字典包含根路径、主记录路径、元数据字段、嵌套字段的处理规则：

config = {
    "root_path": "test",  # 顶层数组的键名
    "main_record_path": "broodje",  # 需要展开的主数组字段
    "meta_fields": ["landje"],  # 需要保留的上层元数据字段
    "nested_fields": {
        # 处理aap嵌套数组：展开并提取内部字段
        "aap": {
            "is_array": True,
            "extract_fields": True
        },
        # 处理days_due单元素数组：直接提取第一个元素
        "days_due": {
            "is_array": True,
            "extract_first": True
        }
    }
}

2. 编写动态处理函数 #

根据配置自动完成嵌套JSON的归一化：

def dynamic_json_normalize(data, config):
    # 获取顶层数据
    root_data = data[config["root_path"]]
    
    # 第一层归一化：展开主记录数组并保留元数据
    df = json_normalize(root_data, record_path=config["main_record_path"], meta=config["meta_fields"])
    
    # 遍历处理所有嵌套字段
    for field, settings in config["nested_fields"].items():
        if field not in df.columns:
            continue
        
        if settings["is_array"]:
            if settings.get("extract_first"):
                # 提取数组第一个元素
                df[field] = df[field].str[0]
            elif settings.get("extract_fields"):
                # 展开嵌套数组并提取内部字段
                nested_df = json_normalize(df[field].explode().tolist())
                df = pd.concat([df.drop(columns=[field]), nested_df], axis=1)
    
    return df

3. 使用动态函数 #

调用函数即可完成动态归一化：

df_dynamic = dynamic_json_normalize(data, config)
# 可选：调整列顺序
df_dynamic = df_dynamic[["landje", "coupon", "type", "dirk_code", "days_due"]]
print(df_dynamic)

后续如果JSON结构变化，只需要修改配置字典，无需修改核心处理代码。

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