Возможно ли сохранить массив данных HASGE dask в паркет?

У меня есть dataframe, состоящий из 100 000+ строк, и каждая строка содержит 100 000 столбцов, всего до 10 000 000 000 значений float.

Мне удалось прочитать их ранее в файле csv (разделенный вкладками), и я успешно прочитал их на 50-ядерном компьютере Xeon с объемом памяти 250 ГБ и попытаюсь записать его как каталог .parq, как например:

Поплавки в huge.csv были сохранены как строки, а это 125 ГБ.

import dask.dataframe as dd
filename = 'huge.csv'
df = dd.read_csv(filename, delimiter='\t', sample=500000000)
df.to_parquet('huge.parq')

Он записывается в huge.parq в течение недели, а каталог составляет 14 ГБ, и похоже, что процесс сохранения .to_parquet не скоро остановится.

И free -mh показывает, что все еще сохраняется память, но время, которое требуется для сохранения каталога .parq, чрезвычайно медленное:

$ free -mh
              total        used        free      shared  buff/cache   available
Mem:           251G         98G         52G         10M        101G        152G
Swap:          238G          0B        238G

Вопросы:

  • Учитывая размер блока данных и машины, возможно ли вообще сохранить файл данных dask в паркетном файле?

  • Нормально ли для dask и fastparquet делать так долго, чтобы сэкономить огромные кадры данных?

  • Есть ли способ оценить время, которое потребуется для сохранения файла паркета?

Ответы

Ответ 1

Как уже говорилось выше, теоретическая причина, по которой .to_parquet() не должна справляться с вашими данными, не существует. Однако количество столбцов чрезвычайно велико, и из-за наличия связанных с ними накладных расходов неудивительно, что процесс занимает много времени - это не типичный вариант использования.

Похоже, ваши данные лучше всего воспринимать как массив, а не таблицу. Существуют механизмы хранения массивов, которые позволяют вам размещать в каждом измерении, например zarr, что также позволяет выполнять различные операции сжатия и предварительной фильтрации которые могут эффективно использовать дисковое пространство. (Другие форматы, такие как HDF5, также популярны для такой задачи)

Пример того, как хранить массив 10k X 10k:

import dask.array as da
import zarr
arr = da.random.random(size=(10000, 10000), chunks=(1000, 1000))
z = zarr.open_array('z.zarr', shape=(10000, 10000), chunks=(1000, 1000), mode='w', dtype='float64')
arr.store(z)

а теперь z.zarr/содержит 100 блоков данных.

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

df = dataframe.read_csv(..)
len(df)  # get length
z = zarr.open_arr(...)  # provide dtype, size and chunk appropriately
df.values.store(z)

или более эффективно обернуть np.loadtxt с помощью dask.delayed, чтобы отказаться от этапа кадра данных.