Может ли dask parralelize читать fm csv файл?

Я конвертирую большой текстовый файл в hdf-хранилище в надежде на быстрый доступ к данным. Преобразование работает в порядке, однако чтение из файла csv не выполняется параллельно. Он очень медленный (занимает около 30 минут для текстового файла 1 ГБ на SSD, поэтому я предполагаю, что он не привязан к IO).

Есть ли способ прочитать его в нескольких потоках в parralel? Sice это может быть важно, я в настоящее время вынужден работать под Windows - на всякий случай, что имеет значение.

from dask import dataframe as ddf
df = ddf.read_csv("data/Measurements*.csv",
             sep=';', 
             parse_dates=["DATETIME"], 
             blocksize=1000000,
             )

df.categorize([ 'Type',
                'Condition',               
          ])

df.to_hdf("data/data.hdf", "Measurements", 'w')

Ответ 1

Да, dask.dataframe может читать параллельно. Однако вы столкнулись с двумя проблемами:

Pandas.read_csv только частично освобождает GIL

По умолчанию dask.dataframe распараллеливается с потоками, потому что большинство панд может работать параллельно в нескольких потоках (освобождает GIL). Pandas.read_csv является исключением, особенно если ваши результирующие кадры данных используют dtypes объекта для текста

dask.dataframe.to_hdf (имя файла) форсирует последовательные вычисления

Запись в один файл HDF приведет к последовательным вычислениям (очень трудно писать в один файл параллельно).

Изменение: новое решение

Сегодня я бы избегал HDF и использовал вместо этого паркет. Я бы, вероятно, использовал многопроцессорные или dask.distributed планировщики, чтобы избежать проблем GIL на одной машине. Сочетание этих двух должно дать вам полное линейное масштабирование.

from dask.distributed import Client
client = Client()

df = dask.dataframe.read_csv(...)
df.to_parquet(...)

Решение

Поскольку ваш набор данных, вероятно, помещается в памяти, используйте dask.dataframe.read_csv для загрузки параллельно с несколькими процессами, а затем немедленно переключитесь на Pandas.

import dask.dataframe as ddf
import dask.multiprocessing

df = ddf.read_csv("data/Measurements*.csv",  # read in parallel
             sep=';', 
             parse_dates=["DATETIME"], 
             blocksize=1000000,
             )

df = df.compute(get=dask.multiprocessing.get)     # convert to pandas

df['Type'] = df['Type'].astype('category')
df['Condition'] = df['Condition'].astype('category')

df.to_hdf('data/data.hdf', 'Measurements', format='table', mode='w')

Ответ 2

В ответ на ответ @MRocklin, в более новых версиях dask вы можете использовать df.compute(scheduler='processes') или df.compute(scheduler='threads') для преобразования в панд с использованием многопроцессорной обработки или многопоточности:

from dask import dataframe as ddf
df = ddf.read_csv("data/Measurements*.csv",
             sep=';', 
             parse_dates=["DATETIME"], 
             blocksize=1000000,
             )

df = df.compute(scheduler='processes')     # convert to pandas

df['Type'] = df['Type'].astype('category')
df['Condition'] = df['Condition'].astype('category')

df.to_hdf('data/data.hdf', 'Measurements', format='table', mode='w')