Использование многопроцессорной обработки для ускорения загрузки панд при разбиении больших файлов CSV - PullRequest
Новая
       15

Использование многопроцессорной обработки для ускорения загрузки панд при разбиении больших файлов CSV

0 голосов
/ 01 апреля 2019

Я пытаюсь прочитать CSV-файл, содержащий 7 миллионов строк и 10 столбцов.Мои аппаратные спецификации следующие:

![enter image description here

Моя стратегия заключается в загрузке наборов данных в виде кусков: 

import pandas as pd
df = pd.read_csv(filename, chunksize = 1000000, low_memory=False)

к сожалению, Я получаю эту ошибку, которая говорит: 

bash:fork:Cannot allocate memory

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

import pandas as pd, numpy as np
from multiprocessing import Pool

def read_csv(filename):
        return pd.read_csv(filename, chunksize = 1000000, low_memory=False)

if __name__ == '__main__':
        pool = Pool(processes = 6)
        df_list = pool.map(read_csv, 'm_datasets.csv.gz')

, но я получаю сообщение об ошибке: 

  File "/opt/rh/rh-python36/root/usr/lib64/python3.6/multiprocessing/pool.py", line 266, in map
    return self._map_async(func, iterable, mapstar, chunksize).get()
  File "/opt/rh/rh-python36/root/usr/lib64/python3.6/multiprocessing/pool.py", line 644, in get
    raise self._value
FileNotFoundError: [Errno 2] File b'm' does not exist: b'm'

при повторном запуске сценариев отображается та же ошибка, но с другой буквой: 

FileNotFoundError: File b'a' does not exist

Что здесь не так?

Я решил вышеупомянутую проблему с помощью apply_async: 

if __name__ == '__main__':
        pool = Pool(processes = 6)
        df_list = pool.apply_async(read_csv,'m_datasets.csv.gz')
        for i in df_list:
                res = i.get()

, однако у меня появилась новая проблема: 

Traceback (most recent call last):

      File "eda.py", line 15, in <module>
        for i in df_list:
    TypeError: 'ApplyResult' object is not iterable

, но когда я использую: 

if __name__ == '__main__':
        jobs = []
        pool = Pool(5)
        lists = pool.apply_async(file_reader, ['m_datasets.csv.gz'])
        for i in lists.get():
                print(i)

Я получаю: 

Traceback (most recent call last):
  File "eda.py", line 12, in <module>
    for i in lists.get():
  File "/opt/rh/rh-python36/root/usr/lib64/python3.6/multiprocessing/pool.py", line 644, in get
    raise self._value
multiprocessing.pool.MaybeEncodingError: Error sending result: '<pandas.io.parsers.TextFileReader object at 0x7f2417ebf748>'. Reason: 'AttributeError("Can't pickle local object '_make_date_converter.<locals>.converter'",)'
...