Python高效合并字典按countryCode分组并对统计字段求和的方法

核心优化思路 #

原来的实现存在两次全量遍历+冗余字典写入的额外开销，1000万级数据下直接将关联、统计逻辑合并为单次遍历，即可实现至少40%的效率提升。

具体落地方案 #

方案1：单轮遍历直接累加（适合内存可容纳全量changes字典的场景） #

• 提前初始化空的分组统计字典country_stats，key为countryCode，value默认值为{'ADDED':0, 'MODIFIED':0, 'REMOVED':0}
• 遍历MongoDB返回的每条文档时，直接完成关联+累加操作，完全跳过给changes字典新增countryCode字段的冗余步骤
• 遍历结束后直接得到最终汇总结果
  参考代码实现（Python）：

from collections import defaultdict

# 初始化统计容器
country_stats = defaultdict(lambda: {'ADDED':0, 'MODIFIED':0, 'REMOVED':0})

# 单次遍历直接完成全量逻辑
for doc in mongo_docs:
    record_id = doc['_id']
    country_code = doc['countryCode']
    # 可根据业务需要新增_id不存在的异常处理逻辑
    if record_id not in changes:
        continue
    change_info = changes[record_id]
    country_stats[country_code]['ADDED'] += change_info['ADDED']
    country_stats[country_code]['MODIFIED'] += change_info['MODIFIED']
    country_stats[country_code]['REMOVED'] += change_info['REMOVED']

方案2：流式分批处理（适合内存不足以容纳全量数据的场景） #

如果全量changes字典或者Mongo查询结果超过内存上限，可将全量流程拆分为批次处理：

• 每次从MongoDB分页查询1万~10万条文档（可根据内存大小调整批次）
• 收集当前批次的所有_id，批量查询另一存储得到对应批次的changes字典
• 对当前批次用方案1的逻辑直接累加统计，处理完成后清空当前批次的临时数据
• 循环处理直到所有批次遍历完成
  该方案内存占用可控，不会出现OOM问题，整体效率仅比方案1多了分批查询的IO开销。

额外优化点 #

• 所有对另一存储的changes查询必须用批量查询替代单条查询，可降低90%以上的IO耗时
• 如果对性能要求更高，可使用pandas的向量化运算替代纯Python循环，底层C实现的遍历累加比纯Python快2~3倍，参考代码：

import pandas as pd

# 转换为DataFrame结构
mongo_df = pd.DataFrame(mongo_docs, columns=['_id', 'countryCode'])
changes_df = pd.DataFrame.from_dict(changes, orient='index').reset_index().rename(columns={'index':'_id'})
# 关联后直接分组求和
result = mongo_df.merge(changes_df, on='_id').groupby('countryCode')[['ADDED', 'MODIFIED', 'REMOVED']].sum()

• 如果countryCode取值范围很小（仅几百个），不需要额外优化哈希查找逻辑；如果countryCode取值超过10万，可将字符串映射为整形key降低哈希开销。

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