Python-Funktionen parallelisieren, ohne Zwischenergebnisse zu speichern

Question

Ich muss die Ergebnisse vieler Bäume für einige Abfragen ansammeln, die ein großes Ergebnis ausgibt. Da alle Bäume unabhängig voneinander behandelt werden können, ist es peinlich parallel, abgesehen von der Tatsache, dass die Ergebnisse summiert werden müssen und ich die Zwischenergebnisse für alle Bäume im Speicher nicht speichern kann. Unten ist ein einfaches Beispiel für einen Code für das Problem, der alle Zwischenergebnisse im Speicher speichert (natürlich sind die Funktionen neuere die gleichen im realen Problem, da dies doppelte Arbeit wäre).

import numpy as np 
from joblib import Parallel, delayed 


functions=[[abs,np.round] for i in range(500)] # Dummy functions 
functions=[function for sublist in functions for function in sublist] 
X=np.random.normal(size=(5,5)) # Dummy data 


def helper_function(function,X=X): 
    return function(X) 
results = Parallel(n_jobs=-1,)(
     map(delayed(helper_function), [functions[i] for i in range(1000)])) 
results_out = np.zeros(results[0].shape) 
for result in results: 
    results_out+=result 

Eine Lösung könnte die folgende Modifikation sein:

import numpy as np 
from joblib import Parallel, delayed 

functions=[[abs,np.round] for i in range(500)] # Dummy functions 
functions=[function for sublist in functions for function in sublist] 
X=np.random.normal(size=(5,5)) # Dummy data 
results_out = np.zeros(results[0].shape) 

def helper_function(function,X=X,results=results_out): 
    result = function(X) 
    results += result 
Parallel(n_jobs=-1,)(
     map(delayed(helper_function), [functions[i] for i in range(1000)])) 

Aber dies könnte Rennen führen. Also ist es nicht optimal.

Haben Sie irgendwelche Vorschläge für das Vorformen, ohne die Zwischenergebnisse zu speichern und trotzdem parallel zu machen?

Answer 1

Die Antwort wird in der documentation of joblib gegeben.

with Parallel(n_jobs=2) as parallel: 
    accumulator = 0. 
    n_iter = 0 
    while accumulator < 1000: 
     results = parallel(delayed(sqrt)(accumulator + i ** 2) 
          for i in range(5)) 
     accumulator += sum(results) # synchronization barrier 
     n_iter += 1 

Sie können die Berechnung in Blöcken tun und die Chunk reduzieren, wie Sie sind aus dem Speicher laufen.