使用Pandas将多级YAML配置扁平化为DataFrame（支持动态层级）

问题背景 #

我可以处理固定层级的YAML配置到DataFrame的转换，但无法适配动态多层级的场景，需要实现一套能兼容任意深度层级、多消息类型（如hero/villain）的转换逻辑，缺失的对应值需显示为null。

示例场景1：固定层级YAML #

原始YAML配置：

content:
  test_group_1:
    segment_1:
        hero:
          '1.0': segment 1 hero msg for 1
          '1.1': segment 1 hero msg for 1.1
    segment_2:
        hero:
          '1.0': segment 2 hero msg for 1
          '1.1': segment 2 hero msg for 1.1
  test_group_2:
    segment_1:
      hero:
        '1.6':  segment 1 hero msg for 1.6
    segment_2:
      hero:
        '1.6': segment 2 hero msg for 1.6

转换后目标DataFrame结构：

核心动态层级需求 #

需兼容任意深度的YAML结构，例如以下场景：

content:
  test_group_1:
    segment_1:
      sub_segment1:
        hero:
          '1.0': sub-segment 1 hero msg for 1
          '1.1': sub-segment 1 hero msg for 1.1
        villain:
          '1.0': sub-segment 1 villain msg for 1
          '1.1': sub-segment 1 villain msg for 1.1
      sub_segment2:
        hero:
          '1.0': sub-segment 2 hero msg for 1
          '1.1': sub-segment 2 hero msg for 1.1
        villain:
          '1.0': sub-segment 2 villain msg for 1

转换后DataFrame需满足：

• 自动生成层级列（如level1/level2/level3，对应层级深度）
• 新增villain列，列名可根据消息节点指定
• 缺失值（如sub_segment2的villain无1.1值）显示为null

解决方案（Python实现） #

使用pyyaml加载YAML，通过递归遍历收集所有层级路径、消息类型、target和对应值，再用pandas整理成结构化DataFrame：

import yaml
import pandas as pd
from collections import defaultdict

def parse_yaml_to_records(data, current_path=None, records=None):
    # 初始化递归变量
    if current_path is None:
        current_path = []
    if records is None:
        records = defaultdict(lambda: defaultdict(dict))
    
    for key, value in data.items():
        if isinstance(value, dict):
            # 判断是否为消息节点：子节点全为字符串类型
            if all(isinstance(v, str) for v in value.values()):
                msg_type = key
                for target, msg in value.items():
                    records[tuple(current_path)][target][msg_type] = msg
            else:
                # 层级节点，递归深入
                parse_yaml_to_records(value, current_path + [key], records)
    return records

# 加载YAML配置（替换为你的文件路径）
with open('config.yaml', 'r', encoding='utf-8') as f:
    config = yaml.safe_load(f)

# 解析YAML为嵌套记录
raw_records = parse_yaml_to_records(config['content'])

# 转换为扁平化行数据
df_rows = []
for path_tuple, target_dict in raw_records.items():
    # 生成层级列（level1、level2...）
    level_cols = {f'level{i+1}': path_tuple[i] for i in range(len(path_tuple))}
    for target, msg_dict in target_dict.items():
        row = {'target': target}
        row.update(level_cols)
        row.update(msg_dict)
        df_rows.append(row)

# 生成DataFrame并填充缺失值
df = pd.DataFrame(df_rows).fillna(value=pd.NA)

# 调整列顺序（target + 层级列 + 消息列）
level_columns = sorted([col for col in df.columns if col.startswith('level')])
other_columns = [col for col in df.columns if col not in level_columns + ['target']]
df = df[['target'] + level_columns + other_columns]

print(df)

代码说明 #

1. 递归遍历：parse_yaml_to_records自动识别层级节点和消息节点，收集所有路径、target与对应消息内容。
2. 扁平化处理：把嵌套的记录转换为每行对应一个target的扁平化结构，自动生成层级列。
3. 缺失值处理：用pd.NA填充缺失的消息值，对应显示为null。
4. 列序优化：将target和层级列前置，提升DataFrame可读性。

动态层级场景输出示例 #

针对第二个YAML示例，输出的DataFrame如下：

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