Dask多分区下map_partitions分组结果重复错误的原因与解决方法

嘿，这个问题我之前用Dask处理聚合操作时也踩过同款坑，咱们一步步拆解清楚：

为什么多分区下用map_partitions会得到重复且不符的结果？ #

你现在用map_partitions对每个分区单独执行groupby.nunique()，本质是让每个分区各自计算自己范围内的分组和唯一值数量，最后Dask只是把所有分区的结果简单拼接起来。这会导致两个核心问题：

1. 重复分组：如果同一个(id, 月份, country, job)分组的数据分散在多个分区里，每个分区都会输出这个分组的一条记录，最终合并后就会出现重复的分组行。
2. 错误的nunique值：每个分区计算的是该分区内source的唯一值数量，而不是全局范围内的唯一值数量，自然和Pandas全局groupby的结果不一致。

当你把分区数设为1时，整个数据集在一个分区里，map_partitions就相当于直接跑Pandas的groupby，所以结果完全正确。

多分区下如何得到正确结果？ #

别用map_partitions做这种需要全局聚合的操作，直接用Dask DataFrame原生的groupby方法就行，Dask会自动处理跨分区的聚合逻辑：

measurements = measurements.repartition(n_partitions=38)
# 用Dask原生groupby替代map_partitions
result = measurements.groupby(
    ["id", measurements.time.dt.to_period("M"), "country", "job"]
).source.nunique().compute().reset_index()

原理说明： #

Dask的groupby会分两步完成全局聚合：

1. 局部聚合：先在每个分区内计算分组的中间结果（比如收集每个分组的source值集合）；
2. 全局合并：把所有分区的中间结果汇总，计算出全局的nunique值，同时保证每个分组只输出一条记录。

这样得到的结果就和Pandas的全局groupby完全一致，还能利用多分区的并行优势提升效率。

额外提醒： #

map_partitions更适合处理每个分区独立完成的任务（比如数据清洗、格式转换、单分区内的统计），而涉及全局聚合的操作（比如groupby、join等），优先用Dask内置的对应方法——它们已经帮你处理好了分区间的协作逻辑。

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