Python, lese viele Dateien und füge die Ergebnisse zusammen

Question

Ich stelle vielleicht eine sehr grundlegende Frage, aber ich kann wirklich nicht herausfinden, wie man eine einfache parallele Anwendung in Python erstellt. Ich führe meine Skripte auf einer Maschine mit 16 Kernen aus und möchte alle effizient nutzen. Ich habe 16 riesige Dateien zu lesen und möchte, dass jeder CPU eine Datei liest und dann das Ergebnis zusammenführt. Hier gebe ich ein schnelles Beispiel dessen, was ich tun möchte:

  parameter1_glob=[]
  parameter2_glob[]

  for cpu in arange(0,16):
      parameter1, parameter2 = loadtxt('file' + str(cpu) + '.dat', unpack=True)

      parameter1_glob.append(parameter1)
      parameter2_glob.append(parameter2)

Ich denke, dass das multiprocessing-Modul helfen könnte, aber ich konnte nicht verstehen, wie man es auf das anwenden soll, was ich tun möchte.

Answer 1

Ich stimme mit dem überein, was Colin Dunklau in seinem Kommentar gesagt hat, dieser Prozess wird beim Lesen und Schreiben dieser Dateien zum Engpass, die CPU-Anforderungen sind minimal. Selbst wenn Sie 17 dedizierte Laufwerke hätten, würden Sie nicht einmal einen Kern voll auslasten. Darüber hinaus, obwohl mir bewusst ist, dass dies tangential zu Ihrer eigentlichen Frage ist, werden Sie wahrscheinlich auf Speicherbeschränkungen mit diesen "riesigen" Dateien stoßen - das Laden von 16 Dateien in den Speicher als Arrays und dann ihr Kombinieren in eine andere Datei wird fast sicher mehr Speicher verbrauchen, als Sie zur Verfügung haben.

Sie können möglicherweise bessere Ergebnisse erzielen, wenn Sie dieses Problem mit Shell-Scripting lösen. Insbesondere verwendet GNU sort einen speichereffizienten Merge-Sort, um sehr schnell eine oder mehrere Dateien zu sortieren - viel schneller als die meisten sorgfältig geschriebenen Anwendungen in Python oder anderen Sprachen.

Ich würde davon abraten, jegliche Art von Multi-Threading-Ansatz zu verwenden, da dies die Komplexität dramatisch erhöht, mit minimalem Nutzen. Stellen Sie sicher, dass Sie so wenig wie möglich der Datei(en) gleichzeitig im Speicher behalten, oder Sie werden schnell keinen Speicher mehr haben. Auf jeden Fall sollten Sie das Lesen und Schreiben unbedingt auf zwei separaten Laufwerken ausführen. Die Verlangsamung, die mit dem gleichzeitigen Lesen und Schreiben auf dieselbe Festplatte verbunden ist, ist äußerst schmerzhaft.

Answer 2

Möchten Sie Zeile für Zeile zusammenführen? Manchmal sind Koroutinen für I/O-gebundene Anwendungen interessanter als klassisches Multitasking. Sie können Generatoren und Koroutinen für alle Arten von Routing, Zusammenführung und Broadcasting verketten. Lassen Sie sich von dieser schönen Präsentation von David Beazley beeindrucken.

Sie können eine Koroutine als Senke verwenden (nicht getestet, bitte beziehen Sie sich auf die Beispiele von dabeaz):

# Eine Senke, die einfach die Zeilen ausgibt
@coroutine
def printer():
    while True:
        line = (yield)
        print line,

quellen = [
    open('file1'),
    open('file2'),
    open('file3'),
    open('file4'),
    open('file5'),
    open('file6'),
    open('file7'),
]

output = printer()
while quellen:
    for quelle in quellen:
        zeile = quelle.next()
        if not zeile: # EOF
            quellen.remove(quelle)
            quelle.close()
            continue
        output.send(zeile)

Answer 3

Vorausgesetzt, dass die Ergebnisse jeder Datei klein sind, können Sie dies mit meinem Paket jug tun:

from jug import TaskGenerator
loadtxt = TaskGenerator(loadtxt)

parameter1_glob=[]
parameter2_glob[]

@TaskGenerator
def write_parameter(oname, ps):
    with open(oname, 'w') as output:
        for p in ps:
            print >>output, p

parameter1_glob = []
parameter2_glob = []

for cpu in arange(0,16):
    ps = loadtxt('file'+str(cpu)+'.dat',unpack=True)
    parameter1_glob.append(ps[0])
    parameter2_glob.append(ps[1])

write_parameter('output1.txt', parameter1_glob)
write_parameter('output2.txt', parameter2_glob)

Jetzt können Sie mehrere jug execute Jobs ausführen.