Wie man multiprocessing pool.map mit mehreren Argumenten verwendet

Question

In der Python multiprocessing Bibliothek, gibt es eine Variante von pool.map die mehrere Argumente unterstützt?

import multiprocessing

text = "test"

def harvester(text, case):
    X = case[0]
    text + str(X)

if __name__ == '__main__':
    pool = multiprocessing.Pool(processes=6)
    case = RAW_DATASET
    pool.map(harvester(text, case), case, 1)
    pool.close()
    pool.join()

Answer 1

Gibt es eine Variante von pool.map, die mehrere Argumente unterstützt?

Python 3.3 enthält pool.starmap() Methode :

#!/usr/bin/env python3
from functools import partial
from itertools import repeat
from multiprocessing import Pool, freeze_support

def func(a, b):
    return a + b

def main():
    a_args = [1,2,3]
    second_arg = 1
    with Pool() as pool:
        L = pool.starmap(func, [(1, 1), (2, 1), (3, 1)])
        M = pool.starmap(func, zip(a_args, repeat(second_arg)))
        N = pool.map(partial(func, b=second_arg), a_args)
        assert L == M == N

if __name__=="__main__":
    freeze_support()
    main()

Für ältere Versionen:

#!/usr/bin/env python2
import itertools
from multiprocessing import Pool, freeze_support

def func(a, b):
    print a, b

def func_star(a_b):
    """Convert `f([1,2])` to `f(1,2)` call."""
    return func(*a_b)

def main():
    pool = Pool()
    a_args = [1,2,3]
    second_arg = 1
    pool.map(func_star, itertools.izip(a_args, itertools.repeat(second_arg)))

if __name__=="__main__":
    freeze_support()
    main()

Ausgabe

1 1
2 1
3 1

Beachten Sie, wie itertools.izip() y itertools.repeat() werden hier verwendet.

Aufgrund von der von @unutbu erwähnte Fehler können Sie nicht verwenden functools.partial() oder ähnliche Fähigkeiten auf Python 2.6, so dass die einfache Wrapper-Funktion func_star() sollte explizit definiert werden. Siehe auch die Abhilfe vorgeschlagen von uptimebox .

Answer 2

Die Antwort auf diese Frage ist versions- und situationsabhängig. Die allgemeinste Antwort für neuere Versionen von Python (seit 3.3) wurde im Folgenden erstmals von J.F. Sebastian . 1 Sie verwendet die Pool.starmap Methode, die eine Folge von Argumenttupeln annimmt. Sie packt dann automatisch die Argumente aus jedem Tupel aus und übergibt sie an die angegebene Funktion:

import multiprocessing
from itertools import product

def merge_names(a, b):
    return '{} & {}'.format(a, b)

if __name__ == '__main__':
    names = ['Brown', 'Wilson', 'Bartlett', 'Rivera', 'Molloy', 'Opie']
    with multiprocessing.Pool(processes=3) as pool:
        results = pool.starmap(merge_names, product(names, repeat=2))
    print(results)

# Output: ['Brown & Brown', 'Brown & Wilson', 'Brown & Bartlett', ...

Für frühere Versionen von Python müssen Sie eine Hilfsfunktion schreiben, um die Argumente explizit zu entpacken. Wenn Sie Folgendes verwenden möchten with müssen Sie auch einen Wrapper schreiben, um die Pool in einen Kontextmanager. (Dank an Myon für den Hinweis darauf).

import multiprocessing
from itertools import product
from contextlib import contextmanager

def merge_names(a, b):
    return '{} & {}'.format(a, b)

def merge_names_unpack(args):
    return merge_names(*args)

@contextmanager
def poolcontext(*args, **kwargs):
    pool = multiprocessing.Pool(*args, **kwargs)
    yield pool
    pool.terminate()

if __name__ == '__main__':
    names = ['Brown', 'Wilson', 'Bartlett', 'Rivera', 'Molloy', 'Opie']
    with poolcontext(processes=3) as pool:
        results = pool.map(merge_names_unpack, product(names, repeat=2))
    print(results)

# Output: ['Brown & Brown', 'Brown & Wilson', 'Brown & Bartlett', ...

In einfacheren Fällen, mit einem festen zweiten Argument, können Sie auch partial aber nur in Python 2.7+.

import multiprocessing
from functools import partial
from contextlib import contextmanager

@contextmanager
def poolcontext(*args, **kwargs):
    pool = multiprocessing.Pool(*args, **kwargs)
    yield pool
    pool.terminate()

def merge_names(a, b):
    return '{} & {}'.format(a, b)

if __name__ == '__main__':
    names = ['Brown', 'Wilson', 'Bartlett', 'Rivera', 'Molloy', 'Opie']
    with poolcontext(processes=3) as pool:
        results = pool.map(partial(merge_names, b='Sons'), names)
    print(results)

# Output: ['Brown & Sons', 'Wilson & Sons', 'Bartlett & Sons', ...

1. Vieles davon wurde durch seine Antwort inspiriert, die wahrscheinlich stattdessen hätte akzeptiert werden sollen. Aber da diese Antwort an der Spitze steht, schien es mir am besten, sie für künftige Leser zu verbessern.

Answer 3

Ich denke, dass das Folgende besser sein wird:

def multi_run_wrapper(args):
   return add(*args)

def add(x,y):
    return x+y

if __name__ == "__main__":
    from multiprocessing import Pool
    pool = Pool(4)
    results = pool.map(multi_run_wrapper,[(1,2),(2,3),(3,4)])
    print results

Ausgabe

[3, 5, 7]

Answer 4

Verwendung von Python 3.3+ con pool.starmap():

from multiprocessing.dummy import Pool as ThreadPool 

def write(i, x):
    print(i, "---", x)

a = ["1","2","3"]
b = ["4","5","6"] 

pool = ThreadPool(2)
pool.starmap(write, zip(a,b)) 
pool.close() 
pool.join()

Ergebnis:

1 --- 4
2 --- 5
3 --- 6

Sie können auch weitere Argumente für zip() angeben, wenn Sie möchten: zip(a,b,c,d,e)

Falls Sie eine konstanter Wert als Argument übergeben:

import itertools

zip(itertools.repeat(constant), a)

Für den Fall, dass Ihre Funktion return etwas:

results = pool.starmap(write, zip(a,b))

Dies ergibt eine Liste mit den zurückgegebenen Werten.

Answer 5

Wie man mehrere Argumente akzeptiert:

def f1(args):
    a, b, c = args[0] , args[1] , args[2]
    return a+b+c

if __name__ == "__main__":
    import multiprocessing
    pool = multiprocessing.Pool(4) 

    result1 = pool.map(f1, [ [1,2,3] ])
    print(result1)