Pyspark/Pandas拆分Map/字典列键并分组聚合累加值实现方法

核心逻辑分三步：拆分Map复合键、分摊对应值、按用户+单键聚合求和后重组Map，以下分别给出PySpark原生实现和Pandas实现。

PySpark 原生实现 #

不需要转Pandas，直接用Spark内置函数处理，性能适配大数据量场景：

import pyspark.sql.functions as F

# 1. 炸开Map列，拿到所有原始复合键与对应计数值
step1 = df.select(
    "CustomerID",
    F.explode("counts").alias("composite_key", "base_cnt")
)

# 2. 按逗号拆分复合键，去除每个键前后多余空格后再次炸开，实现值的分摊
step2 = step1.withColumn(
    "single_category",
    F.explode(F.transform(F.split("composite_key", ","), F.trim))
)

# 3. 按用户ID+拆分后的独立分类分组，累加计数值
step3 = step2.groupBy("CustomerID", "single_category").agg(
    F.sum("base_cnt").alias("total_cnt")
)

# 4. 按用户ID分组，将聚合后的键值对重组为Map结构
final_df = step3.groupBy("CustomerID").agg(
    F.map_from_entries(
        F.collect_list(F.struct("single_category", "total_cnt"))
    ).alias("counts")
)

# 输出结果
final_df.show(truncate=False)

运行后输出和预期完全一致，其中3004000304用户的OTHER STUFF计数为2+2=4，ACCESSORIES计数为1+2=3，其余键值匹配预期。

Pandas 实现 #

适合小数据量场景，逻辑直观易调整：

import pandas as pd
from collections import defaultdict

# 初始化聚合容器
user_category_cnt = defaultdict(lambda: defaultdict(int))

# 遍历每一行数据做拆分和累加
for _, row in df.iterrows():
    cid = row["CustomerID"]
    for composite_key, cnt in row["counts"].items():
        # 拆分复合键，去除前后空格
        for single_key in [k.strip() for k in composite_key.split(",")]:
            user_category_cnt[cid][single_key] += cnt

# 转换为DataFrame格式
final_pd = pd.DataFrame([
    {"CustomerID": cid, "counts": dict(cnt_map)}
    for cid, cnt_map in user_category_cnt.items()
])

print(final_pd)

代码兼容同个用户存在多行counts记录的场景，不需要提前做行级合并，直接遍历即可完成全量累加。

你之前的代码问题在于没有做复合键拆分和值分摊的步骤，直接对原始复合键做了collect_list，自然无法得到单键聚合的结果。

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