ошибка памяти при загрузке dask-данных из сжатых csvs в архив - PullRequest
Новая
       36

ошибка памяти при загрузке dask-данных из сжатых csvs в архив

0 голосов
/ 18 января 2019

Я хотел бы загрузить несколько CSV-файлов из ZIP-архива в DASK. Я попробовал несколько подходов, описанных ниже, но независимо от того, что я пытаюсь, я получаю ошибку памяти, даже когда кажется, что я обошел проблемы, которые, как мне кажется, у меня есть.

Ограничения памяти на диске не позволяют мне распаковывать файлы, поэтому я не могу использовать dd.read_csv (). Поэтому я использую задержку (pd.read_csv ()), затем dd.from_delayed (). Поскольку каждый кадр данных, который я читаю по отдельности, помещается в память, но коллекция не будет, после того, как я прочитал в каждом кадре данных, я перераспределяю его, вероятно, в ложном убеждении, что dask содержит один раздел в памяти за раз. Я вижу, что каждый раздел использует равную память, когда я использую get_partition. Когда я использую memory_usage, даже после перераспределения, использование памяти указывает, что полный фрейм данных находится в памяти, а не один раздел. Поэтому каждый раз, когда я загружаю в csv, используя приведенный ниже код, используя top, я вижу, что моя память постепенно заполняется, пока не получит ошибку памяти. Я также попытался ограничить объем памяти, используемой с set_options, в надежде, что это заставит dask перенести данные из памяти на диск, но это не работает. Я не думаю, что возможно, что дисковое пространство заполняется также. Я думаю, что это RAM. 

import pandas as pd
import zipfile
import dask.dataframe as dd
from dask.delayed import delayed
import dask

dask.set_options(available_memory=3e9)    

with zipfile.ZipFile(path_to_zip_archive) as z:
    with z.open(compressed_csv_1) as f:
        df_1 = delayed(pd.read_csv(f))
        df_1 = dd.from_delayed([df_1], meta=my_dtypes)
        df_1 = df_1.repartition(npartitions=100)
    with z.open(compressed_csv_2) as f:
        df_2 = delayed(pd.read_csv(f))
        df_2 = dd.from_delayed([df_2], meta=my_dtypes)
        df_2 = df_2.repartition(npartitions=100)

с 

df_1.memory_usage()

, чтобы получить использование памяти

и т.д.

Я бы хотел иметь возможность читать csvs из сжатого zip-архива и объединять их для получения одного кадра данных dask без ошибки памяти.

...