Как эффективно переместить файл данных / Dask размером 67 ГБ, не загружая его полностью в память?

Question

Как эффективно переместить файл данных / Dask размером 67 ГБ, не загружая его полностью в память?

Я думаю, что это самый простой способ сделать это, используя JSON stringify, и это может быть лучшим решением в некоторых ситуациях:

JSON.stringify(a1) === JSON.stringify(a2);

Это преобразует объекты a1 и a2 в чтобы их можно было сравнить. Порядок очень важен в большинстве случаев, поскольку он может сортировать объект с помощью алгоритма сортировки, показанного в одном из приведенных выше ответов.

Пожалуйста, обратите внимание, что вы больше не сравниваете объект, а строковое представление объект. Это может быть не совсем то, что вы хотите.

3

python dataframe dask

задан Joe B 16 January 2019 в 01:25

2 ответа

Другие вопросы по тегам:

python dataframe dask

Похожие вопросы:

score 0 · Answer 1

Я бы создал промежуточный файл и использовал бы fp.seek, чтобы записать их в двоичном формате в новом порядке, прежде чем преобразовать его обратно в новый CSV. Если строка, столбец становится столбцом, строка - sys.float_info предоставит вам размер каждого элемента, позицию каждого элемента ((это столбец * old_row_length + строка) * размер с плавающей точкой).

Затем вы заново объединяете их в CSV, конвертируя их обратно в текст и читая в строке old_count_rows.

score 0 · Answer 2

Я изменил свой оригинальный скрипт для развертывания на любом количестве процессоров. Это работало намного быстрее, так как я мог использовать несколько потоков и развертываться на AWS. Я использовал 96-ядерный компьютер, который выполнил задачу примерно за 8 часов. Я был очень удивлен, так как это почти линейное масштабирование! Идея состоит в том, чтобы сделать некоторые повторяющиеся задачи распространяемыми. После этого вы сможете назначать задачи процессору. Здесь распараллеливание выполняется командой pool.map().

Использование этого скрипта из командной строки довольно просто:

python3 transposer.py -i largeFile.tsv

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

import argparse, subprocess
import numpy as np
import pandas as pd
import dask.dataframe as dd
from IPython.display import clear_output
from contextlib import closing
from os import cpu_count
from multiprocessing import Pool

parser = argparse.ArgumentParser(description='Transpose csv')
parser.add_argument('-i', '--infile', help='Path to input folder',
                    default=None)
parser.add_argument('-s', '--sep', help='input separator',
                    default='\t')

args = parser.parse_args()
infile = args.infile
sep = args.sep    
df = pd.read_csv(infile, sep='\t', nrows=3)    

def READ_COL(item):
    print(item)
    outfile = 'outfile{}.temp'.format(item)
    if item !=0:
                x = "awk '{print $"+str(item)+"}' "+infile+" > "+outfile
                subprocess.check_call([x], shell=True)
                col = pd.read_csv(outfile)
                row = col.T
                display(row)
                row.to_csv('col_{:09d}.csv'.format(item), header=False)
                subprocess.check_call(['rm '+outfile], shell=True)
                print(item/len(df.columns))

with closing(Pool(processes=cpu_count())) as pool:
    pool.map(READ_COL, list(range(1, len(df.columns)+1)))