Dask map_partitions приводит к дублированию при уменьшении и дает неверные результаты по сравнению с чистыми pandas

Question

Когда я использую dask для группирования с использованием map_partitions, я получаю дублированные данные и неверные результаты по сравнению с простым pandas groupby. Но когда я использую n_partitons = 1, я получаю правильные результаты. Почему это происходит? и как я могу использовать несколько разделов и по-прежнему получать правильные результаты?

мой код

measurements = measurements.repartition(n_partitions=38)
measurements.map_partitions(lambda df : df.groupby(["id",df.time.dt.to_period("M"), 
"country","job"]).source.nunique()).compute().reset_index()

В pandas я делаю

measurements.groupby(["id",measurements.time.dt.to_period("M"), 
    "country","job"]).source.nunique().reset_index()

PS: Я использую локальный кластер на одной машине.

Answer 1

Когда вы вызываете map_partitions, вы говорите, что хотите выполнить это действие для каждого раздела. Принимая во внимание, что каждое уникальное значение группировки может встречаться в нескольких разделах, вы получите запись для каждой группы, для каждого раздела, в котором оно найдено.

Что если бы был способ сделать групповку между разделами и иметь результаты хитроумно сливаются для вас автоматически? К счастью, это именно то, что делает dask, и вам совсем не нужно было использовать map_partitions.

measurements.groupby(...).field.nunique().compute()