Concat огромные CSV-файлы, используя Dask

Question

Я пытаюсь объединить три файла CSV (8G, 4G, 6G соответственно) в один файл CSV,

, и моя память составляет 16G, есть ли способ объединить эти файлы CSV по столбцам не имея ошибки памяти?

Мои наборы данных похожи на

A  B  C             D   E   F           G    H    I
1  2  3             4   5   6           7    8    9

Моя цель - объединить их в

A  B  C  D  E  F  G  H  I 
  ...

Мой код похож на

def combine_features(raw_feature_dir,connect_feature,time_feature_dir,feature_set):
df1 = dd.read_csv(raw_feature_dir)
df2 = dd.read_csv(connect_feature)
# df3 = dd.read_csv(time_feature_dir)

gc.collect()
df4 = df1.merge(df2)

df4.to_csv(feature_set)

I ' Я планирую сначала объединить два файла, затем объединить следующий, но он все еще показывает ошибку памяти

Есть ли способ объединить огромные CSV-файлы с помощью Dask? или другие инструменты

Например, чтобы сжать файлы CSV, а затем Concat? или использовать генератор, такой как обработчик чтения и записи, который каждый раз берет кусок данных

Спасибо!

Answer 1

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

Вы можете использовать опции skiprows и nrows метода pandas read_csv для чтения в определенном диапазоне c индексов из ваших 3 файлов, объединить в один фрейм данных в памяти (представляющий фрагмент нужного вам CSV), а затем добавьте к CSV, который вы строите на диске.

Другой вариант - использовать другой формат хранения, который может позволить более эффективно добавлять столбцы. В Dask, похоже, есть несколько опций .

Dask также имеет опцию single_file для своего метода to_csv , но я не думаю, что это поможет вам дело, так как вам нужно добавить столбцы.

Answer 2

Я предполагаю, что у вас есть стандартные файлы CSV. Менее трудоемким способом является использование только модуля csv. Таким образом, вы будете обрабатывать по одной строке за раз:

def combine_features(raw_feature_dir,connect_feature,time_feature_dir,feature_set):
    with open(raw_feature_dir) as fd1, open(connect_feature) as fd2, open(time_feature_dir) as fd3,open(feature_set, "w") as fdout:
        fds = [fd1, fd2, fd3]
        readers = [csv.reader(fdi) for fdi in fds]
        writer = csv.writer(fdout)
        try:
            while True:
                row = [field for field in r for r in [next(reader) for reader in readers]]
                writer.writerow(row)
        except StopIteration:
            pass

Осторожно: приведенный выше код предполагает, что:

• все строки во всех входных CSV-файлах верны (без строки с количеством полей, отличным от заголовка одного и того же файла)
• все файлы csv имеют одинаковую длину

Если эти предположения могут быть неверными, код должен:

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

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

Answer 3

Я думаю, вы не хотите использовать слияние, но concat, как указано в вашем вопросе.

Найдите ниже простой пример:

import pandas as pd
import dask.dataframe as dd

df1 = dd.from_pandas(pd.DataFrame({'A':[1,2,3],
                                   'B':[1,4,3], 
                                   'C':[1,2,5]}), 
                                    npartitions=10)
df2 = dd.from_pandas(pd.DataFrame({'D':[0,2,3], 
                                   'E':[1,9,3], 
                                   'F':[1,6,5]}), 
                                    npartitions=10)

dd.concat([df1,df2], axis=1).head(5, npartitions=2)

Вывод:

   A  B  C  D  E  F
0  1  1  1  0  1  1
1  2  4  2  2  9  6
2  3  3  5  3  3  5