Python: das Schreiben auf eine einzige Datei mit dem queue während der Verwendung von multiprocessing Pool

Ich habe Hunderte von tausenden von text-Dateien, die möchte ich analysieren, auf verschiedene Weise. Ich möchte speichern Sie die Ausgabe in eine einzelne Datei ohne Probleme bei der Synchronisierung. Ich habe mit multiprocessing-pool, um dies zu tun, um Zeit zu sparen, aber ich kann nicht herausfinden, wie zu kombinieren, Pool und Warteschlange.

Den folgenden code speichert die infile-name sowie die maximale Anzahl von aufeinander folgenden "x"s in der Datei. Allerdings möchte ich, dass alle Prozesse, Ergebnisse speichern, um die gleiche Datei, und nicht auf verschiedene Dateien, wie in meinem Beispiel. Jede Hilfe auf diesem würde sehr geschätzt werden.

import multiprocessing

with open('infilenamess.txt') as f:
    filenames = f.read().splitlines()

def mp_worker(filename):
 with open(filename, 'r') as f:
      text=f.read()
      m=re.findall("x+", text)
      count=len(max(m, key=len))
      outfile=open(filename+'_results.txt', 'a')
      outfile.write(str(filename)+'|'+str(count)+'\n')
      outfile.close()

def mp_handler():
    p = multiprocessing.Pool(32)
    p.map(mp_worker, filenames)

if __name__ == '__main__':
    mp_handler()
InformationsquelleAutor risraelsen | 2014-10-27
  1. 30

    Multiprocessing-pools implementieren Sie eine Warteschlange für Sie. Nutzen Sie einfach die pool-Methode liefert, dass der Arbeiter Rückgabewert an den Aufrufer. imap funktioniert gut:

    import multiprocessing 
import re

def mp_worker(filename):
    with open(filename) as f:
        text = f.read()
    m = re.findall("x+", text)
    count = len(max(m, key=len))
    return filename, count

def mp_handler():
    p = multiprocessing.Pool(32)
    with open('infilenamess.txt') as f:
        filenames = [line for line in (l.strip() for l in f) if line]
    with open('results.txt', 'w') as f:
        for result in p.imap(mp_worker, filenames):
            # (filename, count) tuples from worker
            f.write('%s: %d\n' % result)

if __name__=='__main__':
    mp_handler()
    • So, ich Schleife durch die Ergebnisse und schreiben Sie Sie zu der Datei, wie Sie kommen? Bedeutet das, dass die neuen Mitarbeiter nicht erst beginnen, wenn jedes "Ergebnis" geschrieben wurde, oder 32 zu einem Zeitpunkt ausgeführt, sondern warten, um zu schreiben? Auch können Sie erklären, warum Sie ersetzt meine f.read().splitlines() mit [line for line in (l.strip() for l in f), wenn Zeile]?
    • Die 32 Prozesse laufen im hintergrund ab und bekommen mehrere Dateinamen in "Blöcken", wie Sie übergeben Ergebnisse an den parent-Prozess. Die Ergebnisse zurückgegeben werden sofort die Eltern tun Ihr parallel arbeiten. Es ist ein wenig effizienter, Lesen Sie die Datei zeilenweise als um das ganze zu Lesen und teilen Sie es später... das ist es, was die Liste ist für.
    • Super Antwort/Beispiel
    • Ausgezeichnet! Es gibt Unmengen von Antworten über dieses online-aber keines so einfach. Dickes Lob an Sie!
    • Vielen Dank! Dies ist nicht ganz klar für mich, dass, Wann und wo das Schwimmbad ist und Wann/wo wird es abgelassen, um die Datei. mein ganzer code ist: Parallel(n_jobs=num_cores)(verzögert(my_function)(entry) für den Eintrag im Inhaltsverzeichnis) es dauert eine Zeile und gibt Sie mit 30 Linien. Muss ich speichern das Ergebnis des ganzen Prozesses in eine Liste und schreiben Sie in die Datei? wenn ja, es ist gonna be so langsam, weil wenn wir irgendwann speichern unsere Ergebnisse in der Liste und es wächst und der Prozess wird so langsam!
    • Wenn die Reihenfolge der zurückgegebenen Ergebnisse nicht anzeigen, Sie können auch imap_unordered.
    • Dies ist eine große Antwort. Ersparte viel Zeit, danke. Ich habe ein problem festgestellt, während dies für meine Daten. Ich habe eine 5GB csv, ich benutze 42-cores zu verarbeiten und schreiben Sie das Ergebnis auf eine einzelne csv - Datei, jedoch gibt es immer ein Durcheinander bei Zeile 147, ich replizieren kann dies über verschiedene sub-Datensätze, aber der gleiche Fehler passiert immer an dieser position.die Reale Produktion cad1d1eaf41c40f89a0198c3be80379f,2018-07-6195d4a2c0914f4381442f08797f658f,2018-06-15 01:47:34,1, gewünschte Ausgabe cad1d1eaf41c40f89a0198c3be80379f,2018-07-30 01:47:34(\n)16195d4a2c0914f4381442f08797f658f,2018-06-15 01:47:34,1

    InformationsquelleAutor tdelaney
  2. 7

    Nahm ich die akzeptierte Antwort und vereinfacht es für mein eigenes Verständnis, wie dies funktioniert. Ich bin Entsendung es hier, falls es jemandem hilft. 

    import multiprocessing

def mp_worker(number):
    number += 1
    return number

def mp_handler():
    p = multiprocessing.Pool(32)
    numbers = list(range(1000))
    with open('results.txt', 'w') as f:
        for result in p.imap(mp_worker, numbers):
            f.write('%d\n' % result)

if __name__=='__main__':
    mp_handler()
    • Vielen Dank! Dies ist nicht ganz klar für mich, dass, Wann und wo das Schwimmbad ist und Wann/wo wird es abgelassen, um die Datei. mein ganzer code ist: Parallel(n_jobs=num_cores)(verzögert(my_function)(entry) für den Eintrag im Inhaltsverzeichnis) es dauert eine Zeile und gibt Sie mit 30 Linien. Muss ich speichern das Ergebnis des ganzen Prozesses in eine Liste und schreiben Sie in die Datei? wenn ja, es ist gonna be so langsam, weil wenn wir irgendwann speichern unsere Ergebnisse in der Liste und es wächst und der Prozess wird so langsam!
    InformationsquelleAutor Raj
Schreibe einen Kommentar