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

Nachdem ich über itertools gelernt habe in J.F. Sebastians Antwort Ich habe beschlossen, einen Schritt weiter zu gehen und einen parmap Paket, das sich um die Parallelisierung kümmert und Folgendes bietet map y starmap Funktionen in Python 2.7 und Python 3.2 (und später auch), die beliebige Nummer von Positionsargumenten.

Einrichtung

pip install parmap

Wie man parallelisiert:

import parmap
# If you want to do:
y = [myfunction(x, argument1, argument2) for x in mylist]
# In parallel:
y = parmap.map(myfunction, mylist, argument1, argument2)

# If you want to do:
z = [myfunction(x, y, argument1, argument2) for (x,y) in mylist]
# In parallel:
z = parmap.starmap(myfunction, mylist, argument1, argument2)

# If you want to do:
listx = [1, 2, 3, 4, 5, 6]
listy = [2, 3, 4, 5, 6, 7]
param = 3.14
param2 = 42
listz = []
for (x, y) in zip(listx, listy):
        listz.append(myfunction(x, y, param1, param2))
# In parallel:
listz = parmap.starmap(myfunction, zip(listx, listy), param1, param2)

Ich habe parmap auf PyPI und auf ein GitHub-Repository .

Die Frage kann zum Beispiel wie folgt beantwortet werden:

import parmap

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

if __name__ == "__main__":
    case = RAW_DATASET  # assuming this is an iterable
    parmap.map(harvester, case, "test", chunksize=1)

Answer 2

Es gibt eine Abzweigung von multiprocessing genannt. Pathos ( Hinweis: Verwenden Sie die Version auf GitHub ), die nicht benötigt werden starmap -- die map-Funktionen spiegeln die API für Pythons map wider, daher kann map mehrere Argumente annehmen.

Con pathos können Sie im Allgemeinen auch Multiprocessing im Interpreter durchführen, anstatt in der __main__ blockieren. Pathos ist nach einigen leichten Aktualisierungen - hauptsächlich der Umstellung auf Python 3.x - für eine Veröffentlichung vorgesehen.

  Python 2.7.5 (default, Sep 30 2013, 20:15:49)
  [GCC 4.2.1 (Apple Inc. build 5566)] on darwin
  Type "help", "copyright", "credits" or "license" for more information.
  >>> def func(a,b):
  ...     print a,b
  ...
  >>>
  >>> from pathos.multiprocessing import ProcessingPool
  >>> pool = ProcessingPool(nodes=4)
  >>> pool.map(func, [1,2,3], [1,1,1])
  1 1
  2 1
  3 1
  [None, None, None]
  >>>
  >>> # also can pickle stuff like lambdas
  >>> result = pool.map(lambda x: x**2, range(10))
  >>> result
  [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
  >>>
  >>> # also does asynchronous map
  >>> result = pool.amap(pow, [1,2,3], [4,5,6])
  >>> result.get()
  [1, 32, 729]
  >>>
  >>> # or can return a map iterator
  >>> result = pool.imap(pow, [1,2,3], [4,5,6])
  >>> result
  <processing.pool.IMapIterator object at 0x110c2ffd0>
  >>> list(result)
  [1, 32, 729]

pathos bietet mehrere Möglichkeiten, um das genaue Verhalten von starmap .

>>> def add(*x):
...   return sum(x)
...
>>> x = [[1,2,3],[4,5,6]]
>>> import pathos
>>> import numpy as np
>>> # use ProcessPool's map and transposing the inputs
>>> pp = pathos.pools.ProcessPool()
>>> pp.map(add, *np.array(x).T)
[6, 15]
>>> # use ProcessPool's map and a lambda to apply the star
>>> pp.map(lambda x: add(*x), x)
[6, 15]
>>> # use a _ProcessPool, which has starmap
>>> _pp = pathos.pools._ProcessPool()
>>> _pp.starmap(add, x)
[6, 15]
>>>

Answer 3

Ein besserer Weg ist die Verwendung einer Tapezierer anstatt eine Wrapper-Funktion von Hand. Besonders wenn Sie viele Funktionen abbilden müssen, wird ein Dekorator Ihre Zeit sparen, da Sie nicht für jede Funktion einen Wrapper schreiben müssen. Normalerweise ist eine dekorierte Funktion nicht picklbar, aber wir können functools um sie zu umgehen. Weitere Diskussionen finden Sie unter aquí .

Hier ist das Beispiel:

def unpack_args(func):
    from functools import wraps
    @wraps(func)
    def wrapper(args):
        if isinstance(args, dict):
            return func(**args)
        else:
            return func(*args)
    return wrapper

@unpack_args
def func(x, y):
    return x + y

Dann können Sie es mit gezippten Argumenten abbilden:

np, xlist, ylist = 2, range(10), range(10)
pool = Pool(np)
res = pool.map(func, zip(xlist, ylist))
pool.close()
pool.join()

Natürlich können Sie jederzeit die Pool.starmap in Python 3 (>=3.3), wie in anderen Antworten erwähnt.

Answer 4

Eine bessere Lösung für Python 2:

from multiprocessing import Pool
def func((i, (a, b))):
    print i, a, b
    return a + b
pool = Pool(3)
pool.map(func, [(0,(1,2)), (1,(2,3)), (2,(3, 4))])

Ausgabe

2 3 4

1 2 3

0 1 2

out[]:

[3, 5, 7]

Answer 5

Eine andere Möglichkeit besteht darin, eine Liste von Listen an eine Ein-Argument-Routine zu übergeben:

import os
from multiprocessing import Pool

def task(args):
    print "PID =", os.getpid(), ", arg1 =", args[0], ", arg2 =", args[1]

pool = Pool()

pool.map(task, [
        [1,2],
        [3,4],
        [5,6],
        [7,8]
    ])

Man kann dann eine Liste von Argumenten mit der von einem bevorzugten Methode konstruieren.