在实际的数据处理工作中，经常会遇到字段值存储为字典格式的字符串的情况，比如从日志、接口返回或者旧系统导出的数据中，这类字符串无法直接参与后续的分析计算，需要将其安全解析后展开为多个独立的列，方便后续使用。本文将详细介绍如何实现这个需求，并且处理各种可能出现的异常情况。
需求背景与场景说明
假设我们有一份用户扩展信息数据，其中扩展属性字段存储的是字典格式的字符串，示例如下：
# 原始数据示例
data = [
{"user_id": 1, "ext_info": "{"age": 25, "city": "北京", "hobby": "跑步"}"},
{"user_id": 2, "ext_info": "{"age": 30, "city": "上海"}"},
{"user_id": 3, "ext_info": "invalid_dict_string"},
{"user_id": 4, "ext_info": "{"age": 28, "city": "广州", "hobby": "阅读", "score": 95}"}
]
我们的目标是将ext_info字段中的字典字符串解析出来，把其中的键作为新的列名，值作为对应列的值，最终得到包含user_id、age、city、hobby、score等多列的结构化数据，同时要保证解析过程安全，不会因无效字符串导致程序崩溃。
安全解析的核心思路
安全解析需要满足以下几个要求：
- 能够识别并跳过无效的字典格式字符串，避免程序抛出异常
- 处理字典中键不统一的情况，缺失的键对应的值填充为默认值
- 支持嵌套字典的解析（如果需要的话），本文先以单层字典为例
核心实现步骤分为三步：第一步是判断字符串是否为有效的字典格式，第二步是解析字符串得到字典对象，第三步是将解析得到的字典展开到原始数据的对应行中。
Python实现方案
1. 基础解析函数实现
首先实现一个安全的字典字符串解析函数，使用json模块来解析，因为标准的字典格式字符串通常符合JSON格式，同时用异常捕获来处理无效字符串的情况：
import json
import pandas as pd
def safe_parse_dict_str(dict_str, default_val=None):
"""
安全解析字典格式字符串
:param dict_str: 待解析的字符串
:param default_val: 解析失败时的返回默认值
:return: 解析后的字典，失败则返回default_val
"""
if not isinstance(dict_str, str):
return default_val
try:
# 尝试解析JSON字符串，兼容双引号格式的字典
parsed = json.loads(dict_str)
# 验证解析结果是否为字典类型
if isinstance(parsed, dict):
return parsed
else:
return default_val
except json.JSONDecodeError:
# 解析失败返回默认值
return default_val
2. 展开为多列结构
接下来结合pandas库实现将解析后的字典展开为多列，首先解析每一行的字典字符串，然后提取所有可能的键作为新列，最后合并数据：
def expand_dict_str_to_columns(df, dict_str_col, default_val=None):
"""
将DataFrame中指定列的字典字符串展开为多列
:param df: 原始DataFrame
:param dict_str_col: 存储字典字符串的列名
:param default_val: 缺失值的填充默认值
:return: 展开后的DataFrame
"""
# 解析每一行的字典字符串
parsed_dicts = df[dict_str_col].apply(lambda x: safe_parse_dict_str(x, {}))
# 收集所有可能的键，用于生成新列
all_keys = set()
for d in parsed_dicts:
all_keys.update(d.keys())
# 为每个键生成新列，缺失的键填充默认值
for key in all_keys:
df[key] = parsed_dicts.apply(lambda x: x.get(key, default_val))
# 可选：删除原始的字典字符串列
# df.drop(columns=[dict_str_col], inplace=True)
return df
# 构造测试DataFrame
df = pd.DataFrame(data)
# 执行展开操作
result_df = expand_dict_str_to_columns(df, "ext_info", default_val=None)
print(result_df)
3. 运行结果说明
上述代码的运行结果如下，可以看到无效的字典字符串对应的新列值填充为None，缺失的键（比如第二行没有hobby和score字段）也自动填充了默认值：
user_id ext_info age city hobby score
0 1 {"age": 25, "city": "北京", "hobby": "跑步"} 25 北京 跑步 NaN
1 2 {"age": 30, "city": "上海"} 30 上海 None NaN
2 3 invalid_dict_string NaN None None NaN
3 4 {"age": 28, "city": "广州", "hobby": "阅读", "score": 95} 28 广州 阅读 95.0
边界情况处理
1. 处理单引号格式的字典字符串
如果字典字符串使用的是单引号（比如Python默认的str(dict)输出格式），json.loads会解析失败，此时可以先做字符串替换，或者使用ast.literal_eval来解析：
import ast
def safe_parse_dict_str_v2(dict_str, default_val=None):
"""支持单引号字典字符串的解析"""
if not isinstance(dict_str, str):
return default_val
try:
# 先尝试JSON解析
parsed = json.loads(dict_str)
if isinstance(parsed, dict):
return parsed
except json.JSONDecodeError:
pass
# JSON解析失败，尝试用ast.literal_eval解析单引号格式的字典
try:
parsed = ast.literal_eval(dict_str)
if isinstance(parsed, dict):
return parsed
except (SyntaxError, ValueError):
pass
return default_val
2. 处理嵌套字典的展开
如果字典中存在嵌套字典，需要递归展开，比如{"a": 1, "b": {"c": 2, "d": 3}}需要展开为a、b_c、b_d三列，可以修改解析和展开逻辑：
def flatten_dict(d, parent_key="", sep="_"):
"""递归展平嵌套字典"""
items = []
for k, v in d.items():
new_key = f"{parent_key}{sep}{k}" if parent_key else k
if isinstance(v, dict):
items.extend(flatten_dict(v, new_key, sep=sep).items())
else:
items.append((new_key, v))
return dict(items)
def expand_nested_dict_str_to_columns(df, dict_str_col, default_val=None):
"""展开嵌套字典字符串为多列"""
parsed_dicts = df[dict_str_col].apply(lambda x: safe_parse_dict_str_v2(x, {}))
# 展平嵌套字典
flattened_dicts = parsed_dicts.apply(lambda x: flatten_dict(x) if isinstance(x, dict) else {})
all_keys = set()
for d in flattened_dicts:
all_keys.update(d.keys())
for key in all_keys:
df[key] = flattened_dicts.apply(lambda x: x.get(key, default_val))
return df
总结
将字典格式字符串安全解析并展开为多列结构的核心是先实现安全的解析函数，捕获所有可能的解析异常，避免程序崩溃，再根据解析得到的字典键动态生成新列，处理键缺失的情况。如果需要处理特殊格式的字符串或者嵌套结构，可以针对性地扩展解析逻辑。这种方法可以广泛应用于数据清洗、日志解析、接口数据转换等场景，提升数据处理的稳定性。