Pandas: Управление большими CSV-файлами: группа + сортировка в новых файлах?

Question

У меня есть очень большой файл CSV для управления, с помощью этого процесса:

• сгруппировать файл по 3 столбцам
• для каждой группы, отсортировать кадр данных по 5 столбцам
• записать этот фрейм данных в CSV-файл

Вот моя первая попытка:

file = pd.read_csv('file.csv')
grouped = file.groupby([col1, col2, col3])
for key, df in grouped: 
    name = 'key.csv'
    df = df.sort_values(by=[col4, col5, col6, col7, col8])
    df.to_csv(name , index=False)
    yield name 

Хорошая идея этого метода: на каждой итерации я могу получить имя файла и поэтому продолжаю процесс ETL для файла, не дожидаясь, пока другой будет готов, и я сортирую непосредственно фрейм данных, прежде чем писать csv.

Плохой момент: файл слишком велик для такой обработки, у меня есть ошибка памяти.

Итак, моя вторая (и текущая) попытка:

list_files = []
for chunk in pd.read_csv('file.csv', chunksize=CHUNKSIZE):
    grouped = chunk.groupby([col1, col2, col3])
    for key, df in grouped:
        name = 'key.csv'
        if Path(name).exists():
            df.to_csv(name, index=False, header=False, mode='a')
        else:
            list_files.append(name)
            df.to_csv(name, index=False)
yield list_files

Здесь: нет проблем с памятью, потому что я читаю файл с чанком.

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

def sort(list_files):
    for filename in list_files:
        df = pd.read_csv(filename)
        df = df.sort_value(..)
        df.to_csv(filename)
        yield filename

Поэтому мне нужно снова прочитать каждый файл, и здесь процесс должен создать все файлы list_files, прежде чем перейти к следующему шагу в процессе ETL

В связи с этим, знаете ли вы, если есть способ (я не вижу его в настоящее время), чтобы решить проблему ошибок памяти и сделать этот процесс группы / сортировать быстрее? Может быть (и, конечно), это невозможно, но любое улучшение поможет (добавьте данные в файл более разумным способом, чем данные уже отсортированы, может быть?)

Спасибо

Редактировать: Может быть, можно было бы отсортировать большой файл, прежде чем читать его, но опять же у меня будет проблема с памятью, не знаю, есть ли другой способ, чем pandas, чтобы сделать это, что будет лучше?

Answer 1

Я был там, и я советую вам использовать Dask, который обеспечивает расширенный параллелизм для аналитики https://dask.org/, несколько похожий на то, что делает Spark. Затем вы можете использовать тот же код, что и при первой попытке

  import dask.dataframe as dd

  file = dd.read_csv('file.csv')
  grouped = file.groupby([col1, col2, col3])
  for key, df in grouped: 
      name = 'key.csv'
      df = df.sort_values(by=[col4, col5, col6, col7, col8])
      df.to_csv(name , index=False)
      yield name 

PS: если при сохранении файла в csv возникает ошибка памяти, используйте параметр chunksize в функции to_csv

Answer 2

Dask реализует большую часть функциональности pandas 'и не даст MemoryError (очевидно, производительность не будет такой блестящей). Аналогичный случай: Killed / MemoryError при создании большого dask.dataframe из отложенной коллекции