Python-Multiprozessing, das sicher in eine Datei schreibt

Ich versuche mich zu lösen, eine große numerische problem, die umfasst eine Vielzahl von Teilproblemen, und ich bin mit Python-multiprocessing-Modul (speziell Pool.Karte) zum aufteilen von verschiedenen unabhängigen Teilprobleme auf verschiedenen Kernen. Jedes subproblem computing umfasst viele sub-Teilproblemen, und ich bin versucht zu effektiv memoize diese Ergebnisse durch Speicherung in einer Datei wenn Sie nicht nochmals berechnet, indem jeder Prozess noch, sonst überspringen Sie die Berechnung und Lesen Sie die Ergebnisse aus der Datei.

Bin ich mit concurrency-Probleme mit den Dateien: verschiedene Prozesse, manchmal zu überprüfen, um zu sehen, wenn ein sub-subproblem wurde berechnet, noch (durch die Suche nach der Datei, wo die Resultate gespeichert werden würde), sehen, dass es nicht, ausführen der Berechnung, dann versuchen Sie, schreiben Sie die Ergebnisse in die gleiche Datei zur gleichen Zeit. Wie kann ich vermeiden, schreiben Kollisionen wie diese?

InformationsquelleAutor der Frage Big Dogg | 2012-11-19

  1. 66

    @GP89 erwähnt eine gute Lösung. Benutzen Sie eine Warteschlange zu senden, die das schreiben von Aufgaben in einem engagierten Prozess ist, der alleinigen Schreibzugriff auf die Datei. Alle anderen Arbeitnehmer haben nur-Lesen-Zugriff. Dadurch wird verhindert Kollisionen. Hier ist ein Beispiel, apply_async, aber es wird funktionieren mit der Karte:

    import multiprocessing as mp
import time

fn = 'c:/temp/temp.txt'

def worker(arg, q):
    '''stupidly simulates long running process'''
    start = time.clock()
    s = 'this is a test'
    txt = s
    for i in xrange(200000):
        txt += s 
    done = time.clock() - start
    with open(fn, 'rb') as f:
        size = len(f.read())
    res = 'Process' + str(arg), str(size), done
    q.put(res)
    return res

def listener(q):
    '''listens for messages on the q, writes to file. '''

    f = open(fn, 'wb') 
    while 1:
        m = q.get()
        if m == 'kill':
            f.write('killed')
            break
        f.write(str(m) + '\n')
        f.flush()
    f.close()

def main():
    #must use Manager queue here, or will not work
    manager = mp.Manager()
    q = manager.Queue()    
    pool = mp.Pool(mp.cpu_count() + 2)

    #put listener to work first
    watcher = pool.apply_async(listener, (q,))

    #fire off workers
    jobs = []
    for i in range(80):
        job = pool.apply_async(worker, (i, q))
        jobs.append(job)

    # collect results from the workers through the pool result queue
    for job in jobs: 
        job.get()

    #now we are done, kill the listener
    q.put('kill')
    pool.close()

if __name__ == "__main__":
   main()

    InformationsquelleAutor der Antwort MikeHunter

  2. 0

    Sieht es für mich aus, dass Sie verwenden müssen Manager vorübergehend speichern Sie Ihre Ergebnisse in eine Liste und schreiben Sie dann die Ergebnisse aus der Liste in eine Datei. Nutzen Sie auch Sternenkarte zu übergeben Sie das Objekt, das Sie Bearbeiten möchten, und die verwalteten Liste. Der erste Schritt ist das erstellen der parameter übergeben werden, die Sternenkarte, die auch die verwalteten Liste.

    from multiprocessing import Manager
from multiprocessing import Pool  
import pandas as pd```

def worker(row, param):
    # do something here and then append it to row
    x = param**2
    row.append(x)

if __name__ == '__main__':
    pool_parameter = [] # list of objects to process
    with Manager() as mgr:
        row = mgr.list([])

        # build list of parameters to send to starmap
        for param in pool_parameter:
            params.append([row,param])

        with Pool() as p:
            p.starmap(worker, params)

    Ab diesem Zeitpunkt müssen Sie entscheiden, wie Sie gehen, um mit der Liste arbeiten. Wenn Sie Tonnen von RAM, und eine riesige Daten-set fühlen Sie sich frei verketten mit pandas. Dann können Sie speichern Sie die Datei ganz einfach als csv-oder eine Gurke.

            df = pd.concat(row, ignore_index=True)

        df.to_pickle('data.pickle')
        df.to_csv('data.csv')

    InformationsquelleAutor der Antwort fizix137

Schreibe einen Kommentar