Dask drop дублирует ошибки памяти split_out

Question

У меня есть большой фрейм данных паркетного даска (40 ГБ) с 600 разделами, и мне нужно drop_duplicates с dask.

Я заметил, что простой drop_duplicates всегда приводит к 1 разделу, поэтому я включил "split_out".

Файл партера с разделами был создан из csvs, каждый из которых уже был дедуплицирован.

Когда я запускаю его, я всегда получаю ошибки памяти рабочих, превышающие 95% памяти.

КогдаНаблюдая за приборной панелью, я также заметил, что работники только заполняют пространство ОЗУ максимально до 70%, поэтому я не понимаю, почему у меня возникают проблемы с памятью.

dataframe.map_partitions(lambda d: d.drop_duplicates('index')) 

.... не будет работать какон выводится только в каждом разделе, но не по всему.

Есть идеи, как рассчитать оптимальный размер раздела, чтобы drop_duplicates работал на моих 2-х работниках с 25 ГБ RAM каждый?

client = Client(n_workers=2, threads_per_worker=2, memory_limit='25000M',diagnostics_port=5001)
b=dd.read_parquet('output/geodata_bodenRaw.parq')
npart = int(b.npartitions)
print('npartitions are: ',npart)
b=b.drop_duplicates(subset='index',split_out=npart)
b=b.map_partitions(lambda d: d.set_index('index'))
b.to_parquet('output/geodata_boden.parq', write_index=True )

Answer 1

Наблюдая за приборной панелью, я также заметил, что работники заполняют пространство ОЗУ максимально до 70%, поэтому я не понимаю, почему у меня возникают проблемы с памятью.

Daskначинает беспокоиться об этом уровне и начинает записывать данные на диск.Он должен иметь немного свободной памяти для выполнения ваших функций.Для получения информации о том, как изменить эти политики, см. Здесь: https://distributed.dask.org/en/latest/worker.html#memory-management

В общем случае уникальность или удаление дубликатов - это дорогостоящая операция, выполняемая параллельно, особенно для наборов данных с большим количеством элементов.Использование параметра split_out - хорошая идея.В вашей ситуации я бы больше с этим играл.