如何按条件聚合DataFrame多列？重写遗留转换脚本遇聚合困境

我明白你在重写遗留转换脚本时，卡在了用户数据的聚合环节——既要把SQL Server里的多对多用户-部门-组数据整合成单用户文档，又要实现DataFrame的多列条件聚合。下面一步步帮你解决这两个核心问题：

一、SQL Server数据聚合为MongoDB用户文档 #

先明确你的SQL表数据：

你的目标是每个用户生成一个文档，包含该用户所有关联的唯一部门和唯一组（注意userid=101的"All users"组重复了，需要去重）。用Pandas处理聚合会比游标逐行处理更高效：

1. 读取SQL数据到DataFrame #

import pandas as pd
import pypyodbc
from pymongo import MongoClient

# 建立SQL连接并读取数据
sql_conn = pypyodbc.connect(sqlConnectionString)
df = pd.read_sql("SELECT userid, dept, groupname, groupid FROM your_table_name", sql_conn)
sql_conn.close()

2. 按用户聚合部门和组 #

这里要处理重复的组数据，确保每个组只在用户文档中出现一次：

# 先对用户+组ID去重，避免同一个组多次出现
unique_groups = df.drop_duplicates(subset=["userid", "groupid"])

# 聚合用户的所有唯一部门
dept_agg = df.groupby("userid")["dept"].unique().reset_index(name="departments")
# 聚合用户的所有唯一组（转为包含groupname和groupid的字典列表）
group_agg = unique_groups.groupby("userid").apply(
    lambda x: x[["groupname", "groupid"]].to_dict("records")
).reset_index(name="groups")

# 合并部门和组的聚合结果，得到每个用户的完整数据
user_combined = pd.merge(dept_agg, group_agg, on="userid", how="inner")

3. 写入MongoDB #

把聚合后的DataFrame转为字典列表，批量插入MongoDB：

# 连接MongoDB并插入数据
mongo_client = MongoClient()
collection = mongo_client.database.collection

# 转为字典列表后批量插入
user_docs = user_combined.to_dict("records")
collection.insert_many(user_docs)

mongo_client.close()

最终生成的MongoDB文档格式会是这样（以userid=101为例）：

{
  "userid": 101,
  "departments": [101, 103, 105],
  "groups": [
    {"groupname": "All users", "groupid": 1001},
    {"groupname": "Admin", "groupid": 1020}
  ]
}

二、DataFrame多列条件分组聚合 #

针对多列条件聚合的需求，我给你几个常见场景的示例：

1. 基础多列分组统计 #

比如按userid和dept分组，统计每个分组下的唯一组数量：

# 多列分组，统计唯一groupid的数量
agg_result = df.groupby(["userid", "dept"])["groupid"].nunique().reset_index(name="unique_group_count")

2. 带过滤条件的聚合 #

比如只处理部门ID≥102的用户，再按userid和groupname分组统计关联的部门数量：

# 先过滤符合条件的数据
filtered_df = df[df["dept"] >= 102]
# 多列分组聚合
conditional_agg = filtered_df.groupby(["userid", "groupname"])["dept"].count().reset_index(name="dept_count")

3. 多字段复杂聚合 #

同时聚合多个指标，比如收集用户的唯一部门列表、统计组的总数、生成组详情列表：

complex_agg = df.groupby("userid").agg(
    unique_departments=("dept", "unique"),
    total_unique_groups=("groupid", "nunique"),
    group_details=("groupid", lambda x: df.loc[x.index, ["groupname", "groupid"]].drop_duplicates().to_dict("records"))
).reset_index()

这样不管是简单统计还是复杂的多维度聚合，都能灵活实现。

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