如何实现Pandas中Lambda与Apply的动态层级分组生成嵌套JSON

问题背景 #

现有一段基于Pandas多列分组生成固定三层嵌套JSON的代码，可满足特定场景需求，但实际业务中分组列数量不固定（可能超过三层），需要实现动态层级的嵌套逻辑，替代原有的硬编码嵌套groupby和apply。

原固定层级代码如下：

test = [df.groupby('cat_a')
          .apply(lambda x: x.groupby('cat_b')
                            .apply(lambda x: [x.groupby('cat_c')
                                               .apply(lambda x: x[['participants_actual','participants_registered']].to_dict('r')
                                                      ).to_dict()]
                                  ).to_dict()
          ).to_dict()]

import json
json_res = list(map(json.dumps, test))

解决方案 #

方法1：递归实现（推荐） #

递归是处理动态嵌套层级最直观的方式，每一层处理一个分组列，完成当前层分组后递归处理剩余列，直到所有分组列处理完毕，最后提取目标列生成叶子节点数据。

完整代码示例：

import pandas as pd
import json

def generate_nested_json(df, group_columns, value_columns):
    def recursive_group(data, cols):
        # 无剩余分组列时，返回目标列的字典列表
        if not cols:
            return data[value_columns].to_dict('records')
        # 按当前列分组，递归处理剩余列构建嵌套结构
        current_col = cols[0]
        return {
            group_key: recursive_group(group_data, cols[1:])
            for group_key, group_data in data.groupby(current_col)
        }
    
    # 生成嵌套结构并转为JSON（与原代码输出格式一致，包裹在列表中）
    nested_result = recursive_group(df, group_columns)
    return json.dumps([nested_result])

# 示例测试
if __name__ == "__main__":
    # 模拟测试数据
    data = {
        'cat_a': ['A', 'A', 'A', 'B', 'B'],
        'cat_b': ['X', 'X', 'Y', 'X', 'Y'],
        'cat_c': ['1', '2', '1', '1', '2'],
        'participants_actual': [10, 20, 15, 5, 25],
        'participants_registered': [15, 25, 20, 10, 30]
    }
    df = pd.DataFrame(data)
    
    # 动态指定分组列（支持任意数量）
    group_cols = ['cat_a', 'cat_b', 'cat_c']
    value_cols = ['participants_actual', 'participants_registered']
    
    json_output = generate_nested_json(df, group_cols, value_cols)
    print(json_output)

代码说明

• 核心函数recursive_group：
  • 终止条件：当没有剩余分组列时，将当前数据的目标列转为字典列表（对应原代码的to_dict('r')）。
  • 递归逻辑：按当前列分组，对每个分组的数据递归处理剩余的分组列，最终构建多层嵌套字典。
• 支持任意数量的分组列，只需修改group_columns参数即可，比如新增cat_d分组列，直接在列表中添加该列名。

方法2：迭代式构建 #

若偏好非递归实现，可通过反向迭代分组列从内到外构建嵌套结构：

import pandas as pd
import json

def generate_nested_json_iterative(df, group_columns, value_columns):
    # 初始化最后一层分组结果
    current_level = df.groupby(group_columns[-1]).apply(lambda x: x[value_columns].to_dict('records')).to_dict()
    
    # 从倒数第二层开始反向遍历分组列，逐层嵌套
    for col in reversed(group_columns[:-1]):
        temp_dict = {}
        # 按当前列分组，将下一层结果嵌套进当前层
        for key, group in df.groupby(col):
            # 获取当前分组对应的下一层键集合
            next_col = group_columns[group_columns.index(col) + 1]
            sub_keys = group[next_col].unique()
            # 筛选并嵌套下一层数据
            temp_dict[key] = {k: current_level[k] for k in sub_keys if k in current_level}
        current_level = temp_dict
    
    return json.dumps([current_level])

这种方法从最内层分组开始，逐步向外层嵌套，逻辑相对递归稍复杂，但同样支持动态分组列数量。

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