Le traitement parallèle d'un grand .fichier csv en Python

Ce qui pourrait être trop tard, mais juste pour les futurs utilisateurs, je vais poster quand même. Une autre affiche mentionné à l'aide de multitraitement. Je peux témoigner de il et peut aller dans le détail. Nous traitons avec des fichiers de plusieurs centaines de MO à plusieurs GO tous les jours à l'aide de Python. Donc, c'est certainement à la hauteur de la tâche. Certains des fichiers nous traitons ne sont pas des CSVs, de sorte que l'analyse peut être assez complexe et plus long que l'accès au disque. Cependant, la méthodologie est la même, peu importe quel type de fichier.

Vous pouvez traiter des pièces de gros fichiers en même temps. Voici le pseudo-code de la façon dont nous le faisons:

import os, multiprocessing as mp
# process file function
def processfile(filename, start=0, stop=0):
if start == 0 and stop == 0:
... process entire file...
else:
with open(file, 'r') as fh:
fh.seek(start)
lines = fh.readlines(stop - start)
... process these lines ...
return results
if __name__ == "__main__":
# get file size and set chuck size
filesize = os.path.getsize(filename)
split_size = 100*1024*1024
# determine if it needs to be split
if filesize > split_size:
# create pool, initialize chunk start location (cursor)
pool = mp.Pool(cpu_count)
cursor = 0
results = []
with open(file, 'r') as fh:
# for every chunk in the file...
for chunk in xrange(filesize // split_size):
# determine where the chunk ends, is it the last one?
if cursor + split_size > filesize:
end = filesize
else:
end = cursor + split_size
# seek to end of chunk and read next line to ensure you 
# pass entire lines to the processfile function
fh.seek(end)
fh.readline()
# get current file location
end = fh.tell()
# add chunk to process pool, save reference to get results
proc = pool.apply_async(processfile, args=[filename, cursor, end])
results.append(proc)
# setup next chunk
cursor = end
# close and wait for pool to finish
pool.close()
pool.join()
# iterate through results
for proc in results:
processfile_result = proc.get()
else:
...process normally...

Comme je l'ai dit, c'est seulement le pseudo-code. Il devrait obtenir n'importe qui a commencé, qui a besoin de faire quelque chose de similaire. Je n'ai pas le code en face de moi, il suffit de le faire à partir de la mémoire.

Mais nous avons eu plus d'une vitesse 2x à partir de cette sur la première manche sans réglage fin. Vous pouvez régler le nombre de processus dans la piscine et la taille des morceaux pour obtenir encore plus accélérer en fonction de votre configuration. Si vous avez plusieurs fichiers, comme nous le faisons, de créer un pool de lire plusieurs fichiers en parallèle. Juste prendre garde à ne pas surcharger la zone avec un trop grand nombre de processus.

Remarque: Vous devez le mettre à l'intérieur un "principal" block pour s'assurer que l'infini les processus ne sont pas créés.

OriginalL'auteur

Si les lignes sont complètement indépendantes juste diviser le fichier d'entrée en autant de fichiers que de Processeurs que vous avez. Après cela, vous pouvez exécuter autant d'instances de processus comme les fichiers d'entrée que vous avez maintenant. Ce cas, car ils sont complètement différents processus, ne sera pas lié par GIL problèmes.

OriginalL'auteur

Viens de trouver une solution à ce problème. J'ai essayé de la Piscine.imap, et il semble que, pour simplifier le traitement de gros fichiers de manière significative. imap a un avantage important quand vient le traitement de gros fichiers: Il renvoie des résultats dès qu'ils sont prêts, et non pas attendre que tous les résultats soient disponibles. Cela permet d'économiser beaucoup de mémoire.

(Ici est non testé l'extrait de code qui lit un fichier csv ligne par ligne, chaque ligne et de l'écrire dans un autre fichier csv. Tout est fait en parallèle.)

import multiprocessing as mp
import csv
CHUNKSIZE = 10000   # Set this to whatever you feel reasonable
def _run_parallel(csvfname, csvoutfname):
with open(csvfname) as csvf, \
open(csvoutfname, 'w') as csvout\
mp.Pool() as p:
reader = csv.reader(csvf)
csvout.writerows(p.imap(process, reader, chunksize=CHUNKSIZE))

OriginalL'auteur