开头做数据处理的人几乎都遇到过这种场景拿到一张表里面有几十万行日志其中一列是一大段 JSON 字符串里面塞着设备信息、来源渠道、用户行为序列。现在想要按用户 ID 分组统计每个人用了什么设备、看了哪些页面、平均停留多久。直接df.groupby(user_id).agg(...)一把梭大概率会报错或者跑出来一团乱麻。这几乎是 Pandas 分组聚合处理 JSON 列时的经典困境。这篇文章想聊透一件事当 DataFrame 里躺着 JSON 列而我们又必须对它做分组聚合时到底怎么处理最高效、最省心。内容会覆盖从解析提速、缓存复用、聚合策略到最终结果整理的完整套路也会把我在实际项目中踩过的坑一并交代清楚。无论你是刚接触 Pandas、还在跟json.loads和apply较劲的新手还是已经处理过不少脏数据、想找更优解法的老手这篇应该都能给你一些可以直接上手的东西。1. 问题拆解JSON 列参与分组聚合痛点到底在哪1.1 三个核心痛点不可哈希、解析开销、聚合语义模糊先别急着写代码我们花两分钟想清楚JSON 列为什么不能像普通数值列那样直接分组第一个痛点是不可哈希。groupby的本质是按照分组键的值做哈希分组所以分组键本身必须是可哈希的。普通的字符串、数值、元组都没问题但如果你的 JSON 列在进入 DataFrame 时已经被json.loads成了 Python 的 dict 或 list那么直接df.groupby(df[extra_info])就会抛TypeError: unhashable type: dict。这个错误几乎所有人都会遇到一次因为很多日志解析脚本读进来之后顺手就把 JSON 列转换成了 dict以为后面会方便结果第一步就卡住了。第二个痛点是解析开销。JSON 列往往是字符串形式存着你要用它就必须先json.loads。如果数据量有几十万行每一行都做一次字符串解析这个开销是非常可观的。更要命的是很多人会在聚合函数里反复解析同一列比如对每个分组都json.loads(x)一次导致同一批数据被解了好几遍白白浪费 CPU 时间。第三个痛点是聚合语义模糊。就算你把 JSON 解析好了面对一个 dict 列你该对它做什么聚合求和求平均都不合适。用户 JSON 里可能有多个字段不同字段的聚合方式完全不同。设备型号可能要取众数停留时长要算均值来源渠道可能要去重拼接。一个单调的agg(sum)根本搞不定你需要的是先把 JSON 拆解成结构化字段再针对每个字段设计独立的聚合逻辑。1.2 两种典型场景观测字段与特征字段在实际项目中JSON 列通常出现在两种不同的语义场景里处理思路也不一样。第一种场景是JSON 作为观测记录。典型例子是埋点日志的附加参数字段里面记录了某次行为发生时的设备型号、网络类型、页面来源等。这类场景的核心诉求是“按某几个 JSON 字段做分组、汇总统计”比如统计不同设备的用户量、不同来源渠道的点击率。处理方式是先把 JSON 里的关键字段拆出来变成独立的 DataFrame 列再做常规的groupby。第二种场景是JSON 作为特征载体。典型例子是用户信息表里有一列tags里面是用户的兴趣标签、历史行为序列。这类场景的核心诉求是“把 JSON 里多个值合并、展开、去重后作为分组结果的一部分”。比如按用户分组把该用户所有行为标签拼成一个去重后的集合。处理方式往往需要自定义聚合函数或者在groupby之前先用explode把数组展开。这两种场景在实际中经常混着出现。我的建议是拿到数据之后先别急着写聚合先花几分钟看下 JSON 列里的数据形态是单个对象还是对象数组字段是固定不变的还是会随事件类型变化想清楚用途才能选对处理路径。1.3 处理思路的整体设计先拆解再聚合基于上述分析我个人总结出的核心思路就一句话尽量在 groupby 之前把 JSON 拆解成普通列拆不了的再用自定义聚合函数兜底。这个思路背后的逻辑很简单。Pandas 的groupby和agg是高度优化过的 C 实现处理普通的数值列和字符串列非常快而自定义 Python 函数走的是解释器循环速度完全不是一个量级。如果能在分组之前把 JSON 里需要统计的字段都提取成普通列后面就能用最快的路径完成聚合。那什么时候不能拆解比如 JSON 长度为 0、字段不固定、字段值本身是嵌套结构拆出来的列会非常稀疏浪费内存和计算资源。这种情况下我倾向于保留 JSON 整体在聚合后把同一分组的 JSON 收集成列表再按需处理。这种方案会在第 3 节详细展开。2. JSON 列解析效率的优化手法2.1 批量解析json.loads 循环的性能瓶颈解析 JSON 最快的方式永远不是 Pandas 的处理而是 Python 标准库json里的 C 加速模块。但很多人写代码的时候会用df[col].apply(json.loads)这其实不是最优选择。我实测过几次在数据量达到几十万行时apply单遍解析本身的耗时还能接受真正的问题出现在两个地方一个是解析结果被反复使用另一个是解析过程中混入了大量异常处理逻辑导致不能走快速路径。先说批量解析的正确打开方式用列表推导式配合json.loads会比apply快不止一点import json import pandas as pd df[parsed] [json.loads(x) if isinstance(x, str) else x for x in df[json_col]]为什么列表推导式更快因为它避免了apply的函数调用开销底层直接走 Python 的 C 循环处理列表省掉了一层层 pandas 封装。实测下来对于 10 万行数据这段代码大约比df[json_col].apply(json.loads)快 15% 到 25%数据量越大差异越明显。更激进的做法是用mapdf[parsed] list(map(json.loads, df[json_col]))这个写法把循环下沉到 C 层理论上更快。但注意如果列里混有空值或已经解析过的 dict就很容易报错所以实际项目中我还是更推荐带条件判断的列表推导式稳一点。2.2 解析结果的缓存复用避免重复解析我见过最多的性能浪费是在不同的聚合步骤里反复解析同一个 JSON 列。有的人写了三个自定义聚合函数每个函数里都json.loads一次10 万行数据就被解析了 3 次耗时直接翻三倍。正确的做法是解析一次把结果缓存在一个单独的列里后续所有聚合都从缓存列取值。json_cache {} def safe_loads(x): if isinstance(x, str): if x not in json_cache: json_cache[x] json.loads(x) return json_cache[x] return x df[parsed] df[json_col].map(safe_loads)这里的json_cache用字符串原文做 key相同内容的 JSON 不会重复解析。业务日志里经常有很多重复或相近的记录这样能省掉大量重复解析时间。在多进程场景下你甚至可以把json_cache序列化到内存里共享进一步提速。2.3 处理 JSON 数组列explode 与逐元素解析上面讨论的是 JSON 字符串是单个对象的情况。一旦 JSON 内容是数组事情就复杂一些了。比如一列的值是[{event: click, ts: 123}, {event: view, ts: 456}]你想按用户分组统计所有事件出现的次数直接解析成列表还不够得把数组拆开。第一种拆法是用explodedf_exploded df.assign(parseddf[json_col].map(json.loads)).explode(parsed) # 此时 parsed 列每个元素都是单个 JSON 对象explode会把每个列表元素展开成一行原来的其他列会自动复制非常方便。展开之后每个对象里的事件名就能用get方法提取。如果 JSON 数组里套数组比如嵌套的items列表可以先explode第一层再对这层里的列表字段做第二次explode。第二种拆法是逐元素解析适合你想保留每个数组元素的原始字符串、但又想快速提取某个字段。这种情况我一般不用explode而是直接在原始 DataFrame 上对每个元素做处理df[first_event] df[json_col].map(lambda s: json.loads(s)[0].get(event))这是典型的“无需全展开、只取关键字段”的场景性能比explode好很多因为完全不产生行数膨胀。2.4 非标准 JSON 的兜底处理现实里的 JSON 字段往往不是标准的。最常见的几个坑包括字符串用了单引号而不是双引号、末尾有多余逗号、里面有裸的NaN或None、整个值其实是 Python dict 打印出来的字符串。应对这些情况有三板斧。第一板斧先pd.json_normalize配合errorsignore看看能不能直接展开。pd.json_normalize是一个很有用的工具能自动把嵌套 JSON 展开成扁平表格但前提是 JSON 对象结构要标准。第二板斧对非标准字符串做预处理。比如把单引号替换成双引号、去掉尾逗号import re def fix_json_string(s): s re.sub(r, , s) s re.sub(r,\s*([}\]]), r\1, s) return s df[json_col_fixed] df[json_col].map(fix_json_string)但注意粗暴替换单引号很危险因为 JSON 字符串值里可能本来就有单引号。这种场景我更推荐用ast.literal_eval它能安全地把 Python 字面量包括 dict、list、字符串转换成对象兼容单引号import ast df[parsed] df[json_col].map(lambda s: ast.literal_eval(s) if isinstance(s, str) else s)第三板斧解析失败时给默认值。我在实际项目中习惯写一个统一的解析函数失败时返回None或空 dict然后单独记录解析失败的行数方便事后排查。3. groupby 聚合阶段对 JSON 列的处理策略3.1 聚合前提取特征字段最推荐的路径在绝大多数场景下最省事、最高效的办法就是先提取、后聚合。比如你的 JSON 列里有device、source、duration三个字段先把这个三个字段批量展开到新列import json def extract_fields(row): parsed row if isinstance(row, dict) else json.loads(row) return pd.Series({ device: parsed.get(device), source: parsed.get(source), duration: parsed.get(duration) }) df[[device, source, duration]] df[json_col].apply(extract_fields)然后就能做非常常规的分组聚合了result df.groupby([user_id, device]).agg( total_duration(duration, sum), visit_count(duration, count), source_set(source, lambda s: ,.join(sorted(set(s)))) ).reset_index()这条路线的最大优势是聚合走 Pandas 的快速路径自定义函数只用在真正需要特殊逻辑的字段上整体性能非常好。如果你只需要 2 到 3 个字段却把整个 JSON 全展开成 DataFrame会带来不必要的列膨胀。所以最佳实践是按需提取而不是全量展开。3.2 聚合时保留 JSON 原始内容有些场景下你不想拆开 JSON而是希望聚合后把每个分组的所有 JSON 串拼接起来。比如一个用户多次访问每次访问都带了不同的特征参数你想在聚合结果里看到这个用户所有特征参数的完整记录。这时候可以在agg里用list、first、last直接收集result df.groupby(user_id, sortFalse).agg( all_jsons(json_col, list), first_json(json_col, first), last_json(json_col, last) ).reset_index()注意list聚合函数会把整个 Series 转成列表如果某个用户的记录数很多结果列表会很长。更稳妥的做法是先做降维比如同一用户对同一 JSON 去重后再收集。这也是我在做用户标签聚合时经常遇到的坑数据重复了标签也跟着膨胀。3.3 自定义聚合函数与 named aggregation如果你确实需要在聚合函数里动态解析 JSON就必须把自定义函数写得尽量高效。这几点经验我想分享一下。第一自定义函数接收的参数是一个 Series访问它的值可以直接走循环但尽量别在循环里json.loads。如果 JSON 列还没解析可以先在分组前统一解析好聚合函数只负责取值不做字符串解析。第二能用named aggregation就不要写apply。比如你想同时计算 JSON 里duration的总和和最大值用agg加多个命名列比单独写一个apply清晰得多result df.groupby(user_id).agg( total_duration(parsed, lambda s: sum(d.get(duration, 0) for d in s if isinstance(d, dict))), max_duration(parsed, lambda s: max((d.get(duration, 0) for d in s if isinstance(d, dict)), default0)) ).reset_index()第三如果你的聚合函数要返回多个字段不要用apply返回 DataFrame那样速度极慢。直接拆成多个agg列或者用groupby.agg的字典格式Pandas 会在 C 层尽量做优化。3.4 聚合结果的扁平化与导出分组聚合跑完之后结果里经常还会残留一些 JSON 结构比如列表类型的all_jsons。为了方便后续写库或生成报表最好统一做一次扁平化。主要的处理思路把 list 列转成以分号分隔的字符串或者转成用json.dumps序列化过的字符串。前者适合直接在 Excel 里看后者适合存数据库。result[all_jsons_summary] result[all_jsons].apply(lambda lst: ; .join(map(str, lst))) result[all_jsons_json] result[all_jsons].apply(lambda lst: json.dumps(lst, ensure_asciiFalse))导出时把不必要的中间列删掉尤其是原始 JSON 列和解析后的 dict 列它们会让 DataFrame 非常大、拖慢 to_csv 或 to_parquet 的速度。4. 一个完整的实操案例用户行为日志的分组统计4.1 场景设定与造数纸上谈兵没意思直接上一个我平时工作里常遇到的场景用户行为日志表每行包含user_id、event_time、extra_info三列其中extra_info是 JSON 字符串结构大概是{ device: iPhone 15 Pro, os: iOS 17.4, source: homepage_banner, duration: 32, tags: [video, recommend] }现在我们要按user_id分组统计每个用户的访问次数、总停留时长、使用过的设备集合、以及所有出现过的标签集合。为了演示效果我直接构造 5 万行模拟数据import json import random import pandas as pd random.seed(42) users [fU{i:05d} for i in range(2000)] devices [iPhone 15 Pro, iPhone 14, Android Pixel 8, iPad Pro, MacBook Air, Huawei Mate 60] sources [homepage_banner, search_recommend, push_notification, external_link] tag_pool [video, recommend, electronics, sports, fashion, music, travel] rows [] for _ in range(50000): user_id random.choice(users) event_time pd.Timestamp(2026-01-01 00:00:00) pd.Timedelta(secondsrandom.randint(0, 30 * 24 * 3600)) extra { device: random.choice(devices), os: iOS 17.4 if iPhone in extra.get(device, ) or random.random() 0.5 else Android 14, source: random.choice(sources), duration: random.randint(1, 300), tags: random.sample(tag_pool, krandom.randint(1, 4)) } rows.append({ user_id: user_id, event_time: event_time, extra_info: json.dumps(extra, ensure_asciiFalse) }) df pd.DataFrame(rows) df.head()这里我故意把os字段写得很随意实际数据里这种不一致非常常见后面会体现出来。4.2 解析与预处理首先做一次性解析并缓存结果json_cache {} def safe_loads(x): if isinstance(x, str): if x not in json_cache: json_cache[x] json.loads(x) return json_cache[x] return x df[parsed] df[extra_info].map(safe_loads)然后按需提取字段。注意tags是一个数组暂时先不展开等聚合时再处理df[device] df[parsed].apply(lambda d: d.get(device) if isinstance(d, dict) else None) df[source] df[parsed].apply(lambda d: d.get(source) if isinstance(d, dict) else None) df[duration] df[parsed].apply(lambda d: d.get(duration, 0) if isinstance(d, dict) else 0)对于tags我建议做一次爆炸让每行变成一个标签df_tags df.explode(tags) df_tags[tag] df_tags[tags]注意这里explode之后如果tags是空列表那一行的tag会是None可以在后续聚合时过滤掉。4.3 分组聚合与结果整理先做最常规的聚合按用户统计访问次数、总停留时长、使用过的设备集合、来源渠道集合。result df.groupby(user_id).agg( visit_count(event_time, count), total_duration(duration, sum), avg_duration(duration, mean), device_set(device, lambda s: ,.join(sorted(set(s)))), source_set(source, lambda s: ,.join(sorted(set(s)))) ).reset_index()再对标签做分组统计统计每个用户的标签集合和每个标签的曝光次数。标签的曝光次数可以直接在df_tags分组tag_summary df_tags[df_tags[tag].notna()].groupby(user_id)[tag].agg( tag_countcount, tag_setlambda s: ,.join(sorted(set(s))) ).reset_index()最后把两个结果合并起来final_result result.merge(tag_summary, onuser_id, howleft)这样一步到位拿到了每个用户的访问次数、停留时长、设备、来源、标签信息。整个过程大概几百毫秒5 万行数据毫无压力。如果你还想进一步验证结果可以抽样几个人看原始数据对比一下这一步在调试阶段尤其重要能帮你发现解析或者聚合逻辑的问题。5. 常见报错与排查技巧5.1 典型报错对照速查表JSON 列加 groupby 最常见的错误几乎都能对应到几个固定原因。我整理了以下速查表遇到问题可以直接对号入座。报错信息常见原因解决方法TypeError: unhashable type: dict/list分组键或聚合列里是 dict/list 对象先转为 JSON 字符串或提取所需标量字段JSONDecodeError: Expecting value字符串是空的、包含 NaN/None/单引号用ast.literal_eval或先预处理KeyError: xxxJSON 对象里没有这个字段用.get(xxx)代替[xxx]或统一填默认值ValueError: cannot reindex from a duplicate axisapply返回 DataFrame 时索引不对齐改用df[[a,b]] df[col].apply(lambda r: pd.Series(...))时确保索引重置聚合结果行数暴增groupby前误用了fillna或对字典列误操作产生额外行拆解前检查dtypes确认目标列是 object 且形态正常结果里出现NaN而不是预期值JSON 里字段缺失或类型不对解析后统一用.get加默认值5.2 性能排查解析慢、内存膨胀如果处理 100 万行数据时中间老是卡住先别急着换库从两个方向排查。第一是检查解析是否发生了重复。最直接的方法是在safe_loads函数里加个计数器统计实际发生了多少次json.loads。如果解析次数远大于总行数说明有重复解析。我发现很多人的apply似乎把函数应用了多次尤其是链式操作或分组后用了多个聚合函数这种情况用缓存列能立刻解决。第二是检查数据里是不是藏了大量重复的 JSON 字符串。如果不同行里 JSON 原文完全相同缓存命中率会非常高解析开销可以忽略。这个特性被很多人忽略因为大家总觉得日志数据各不相同。实际上埋点日志里大量用户的行为组合就是那么几种JSON 字符串重复率相当高。善用缓存字典性能提升立竿见影。内存膨胀又是一个容易被忽略的点。df[parsed] df[extra_info].map(json.loads)之后parsed列里存的是一个个 Python dict 对象内存开销远大于原始字符串。如果数据量特别大建议提取完字段后立刻删掉parsed列df.drop(columns[parsed, extra_info], inplaceTrue)5.3 数据质量排查非法 JSON 与 dtype 陷阱5.3.1 非法 JSON 的静默处理我强烈建议在解析阶段就统计失败数量而不是等到聚合完才发现数据对不上。用一个带错误计数的解析函数parse_errors 0 def safe_loads_with_count(x): global parse_errors if isinstance(x, str): try: return json.loads(x) except Exception: parse_errors 1 return None return x df[parsed] df[extra_info].map(safe_loads_with_count) print(解析失败行数, parse_errors)如果解析失败行数占比较高比如超过 0.1%就别直接往下走了先人工看一眼失败样例确定是格式问题还是字段类型问题再决定是清洗还是过滤。静默丢弃是最危险的做法因为它会让聚合结果莫名其妙地和预想不符。5.3.2 dtype 陷阱object 列里的 Str 和 Dict 混着一个常见的坑是DataFrame 里某列名义上叫extra_info实际上里面既有字符串形式的 JSON又有已解析过的 dict甚至还有None或空字符串。这通常是因为数据从多个数据源拼接而来或者读库时某个接口已经做过一次解析。这种混合类型会导致两个问题一是df[extra_info].map(json.loads)直接报错二是groupby时出现 unhashable。正确做法是先统一类型把 dict 和 str 都转成 JSON 字符串或者统一转成 dictdef unify_json_cell(x): if isinstance(x, str): return x if isinstance(x, (dict, list)): return json.dumps(x, ensure_asciiFalse) return df[extra_info_unified] df[extra_info].map(unify_json_cell)之后再统一解析就不会再遇到类型混乱的问题。5.3.3 聚合结果出现多层嵌套 vs 扁平化最后提醒一下agg返回的列表list列如果直接写入 CSV会被自动转成字符串表示比如[video, travel]后面再读回来就是一个没有引号的字符串很麻烦。所以聚合结果在导出前一定要考虑好格式。我现在的习惯是需要入库的用json.dumps序列化需要进 Excel 给业务方看的用分号分隔字符串需要做二次分析的保留 DataFrame 的列结构不要提前转成字符串。6. 从实战角度再补充几个小技巧除了上面讲到的方案还有几个零散但很实用的小技巧一并分享出来。第一如果 JSON 列的结构非常标准且字段固定可以优先试试pd.json_normalize。它能一次性把整个 JSON 列展开成多个列速度也不慢。但注意它不能直接处理“列里嵌套数组”的情况数组部分要单独处理。所以我说它适合简单的对象型 JSON不适合复杂的数组型 JSON。第二处理超大 DataFrame 时可以尝试用polars或duckdb做同类操作它们的 JSON 解析和分组聚合在性能上可能比 Pandas 好很多。但考虑到很多团队现有代码都是 Pandas 生态我不会在本文展开讲其他库只是在项目选型时提醒一句不要让 Pandas 成为唯一选项。第三调试的时候别拿全量数据一次次跑。先df.sample(1000)或者df.head(1000)做小样本验证逻辑通了再全量跑。这个习惯能省下大量时间尤其是处理日志数据的场景。第四善用groupby(..., sortFalse)。如果你不需要按分组键排序显式传sortFalse能省去一次排序的开销。数据量大时这是一个很可观的性能提升。我第一次注意到这个参数是在处理包含几十万个用户的日志时加上sortFalse之后聚合耗时大约降了三分之一。第五如果 JSON 字段非常多建议把提取逻辑封装成一个可复用的函数。后续数据源结构变了只改这一个函数就行不用到处改代码。结尾实际项目里JSON 列的分组聚合处理往往不会一帆风顺数据里藏着各种意想不到的格式问题。我自己最有体会的一点是先把 JSON 列变成普通列后面的事就好办太多了。很多看起来复杂的需求拆解后都不过是普通的分组、聚合、合并。反过来如果一开始就硬碰硬地对 JSON 列做复杂操作最后只能不停地和各种异常作斗争。最后再分享一个小建议写 JSON 解析和分组聚合的代码时记得在关键节点打印几条中间结果的样本特别是解析完和聚合完这两个节点。这些看起来不起眼的检查往往会帮你提前发现数据质量问题省下大量的返工时间。希望这篇文章能帮你少踩几个坑。