遍历嵌套数据避免内层循环（性能优化）：百万级数据集ID真值计数提速

问题背景 #

处理百万行以上的大型数据集，需要统计每个ID对应的report=true的数量，并生成指定格式的字典。现有代码可实现功能，但性能极差，需优化。

原始数据结构 #

{
  "data": {
    "employees": [
      {
        "id": 274,
        "report": true
      },
      {
        "id": 274,
        "report": false
      },
      {
        "id": 276,
        "report": true
      },
      {
        "id": 276,
        "report": true
      },
      {
        "id": 278,
        "report": true
      },
      {
        "id": 278,
        "report": false
      }
    ]
  }
}

期望输出格式 #

{274: {'id': 274, 'count': 1}, 276: {'id': 276, 'count': 2}, 278: {'id': 278, 'count': 1}}

现有代码的问题 #

当前代码存在两个致命问题：

1. 时间复杂度极高：每次遇到新ID时，都要遍历整个数据集进行分组，时间复杂度为O(n²)，百万级数据下会导致运行时间爆炸。
2. 逻辑错误：循环内部直接return final_dict，会导致只处理第一个ID就终止程序，根本无法完成全量数据的统计。

final_dict = {}

for employee in result["data"]["employees"]:
    if employee["id"] not in final_dict.keys():
        final_dict[employee["id"]] = {"id": employee["id"]}
        grouped_results = [res for res in result["data"]["employees"] if
                           employee["id"] == res['id']]
        final_dict[employee["id"]]["count"] = len(
            [res for res in grouped_results if res["report"]]
        )

        return final_dict  # 此处错误：提前返回，未处理全量数据

优化方案 #

方案1：单次遍历统计（最优，O(n)时间复杂度） #

只遍历数据集一次，每个元素仅处理一次，彻底避免多层循环。

final_dict = {}
for employee in result["data"]["employees"]:
    emp_id = employee["id"]
    # 初始化ID对应的统计结构（仅第一次遇到时执行）
    if emp_id not in final_dict:
        final_dict[emp_id] = {"id": emp_id, "count": 0}
    # 如果report为True，计数加1
    if employee["report"]:
        final_dict[emp_id]["count"] += 1

方案2：用collections.defaultdict简化代码 #

逻辑和方案1一致，只是用defaultdict简化初始化操作：

from collections import defaultdict

final_dict = defaultdict(lambda: {"id": None, "count": 0})
for employee in result["data"]["employees"]:
    emp_id = employee["id"]
    entry = final_dict[emp_id]
    # 第一次赋值ID
    if entry["id"] is None:
        entry["id"] = emp_id
    # 累加truthy计数
    if employee["report"]:
        entry["count"] += 1
# 按需转换为普通字典
final_dict = dict(final_dict)

方案3：Pandas向量化处理（超大规模数据首选） #

对于百万级甚至更大的数据集，Pandas的C级向量化操作比纯Python循环效率更高：

import pandas as pd

# 将数据集转换为DataFrame
df = pd.DataFrame(result["data"]["employees"])
# 按ID分组，统计report为True的数量
count_df = df[df["report"]].groupby("id").size().reset_index(name="count")
# 转换为目标格式的字典
final_dict = count_df.set_index("id").apply(lambda x: x.to_dict(), axis=1).to_dict()

# 可选：补充report全为False的ID，计数设为0
all_ids = df["id"].unique()
for emp_id in all_ids:
    if emp_id not in final_dict:
        final_dict[emp_id] = {"id": emp_id, "count": 0}

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