Python çoklu işleme güvenli bir şekilde bir dosyaya yazıyor

Question

Çok sayıda alt problem içeren büyük bir sayısal problemi çözmeye çalışıyorum ve farklı çekirdekler üzerine farklı bağımsız alt problemleri ayırmak için Python'un çoklu işlem modülünü (özellikle Pool.map) kullanıyorum. Her alt problem bir çok alt-alt-problemin hesaplanmasını içerir ve bu sonuçları henüz herhangi bir işlemle hesaplanmamışlarsa bir dosyaya saklayarak etkin bir şekilde belleğe almaya çalışıyorum, aksi halde hesaplamayı atlayın ve sadece sonuçları dosyadan okuyun.

Dosyaları ile eşzamanlılık sorunları yaşıyorum: farklı işlemler bazen bir alt-altproblemin hesaplanmış olup olmadığını (sonuçların saklanacağı dosyaya bakarak) kontrol edip, olmadığını görüyorlar. hesaplamayı çalıştırın, ardından sonuçları aynı dosyaya aynı anda yazmaya çalışın. Böyle çarpışmalar yazmayı nasıl önleyebilirim?

Answer 1

@ GP89 iyi bir çözümden bahsetmiştir. Yazma görevlerini, dosyaya tek yazma erişimi olan özel bir işleme göndermek için bir sıra kullanın. Diğer tüm çalışanlar sadece erişimi okudu. Bu çarpışmaları ortadan kaldıracaktır. İşte apply_async kullanan bir örnektir, ama çok harita ile çalışacaktır:

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() 

iyi şanslar,

Mike

Bu geçici sonuçlarını kaydetmek için Manager'ı kullanmak gerektiğini bana bakıyor

Answer 2

bir listeye ve sonuçları listeden bir dosyaya yazın. Ayrıca, işlemek istediğiniz nesneyi ve yönetilen listeyi geçmek için starmap kullanın. İlk adım, yönetilen listeyi içeren starmap'e geçirilecek parametrenin oluşturulmasıdır.

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) 

Bu noktadan, listeyi nasıl ele alacağınıza karar vermeniz gerekir. Eğer tonlarca RAM'iniz varsa ve büyük bir veri seti pandaları kullanarak birleştirme konusunda çekinmeyin. Ardından, dosyayı csv veya turşu olarak kolayca kaydedebilirsiniz.

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

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