Czy w bibliotece wieloprocesowej Python istnieje wariant pool.map, który obsługuje wiele argumentów?

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

Odpowiedź na to pytanie zależy od wersji i sytuacji. Najbardziej ogólną odpowiedź na ostatnie wersje Pythona (od 3.3) po raz pierwszy opisano poniżej JF Sebastian . ¹ Wykorzystuje Pool.starmapmetodę, która akceptuje ciąg krotek argumentów. Następnie automatycznie rozpakowuje argumenty z każdej krotki i przekazuje je do podanej funkcji:

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', ...

We wcześniejszych wersjach Pythona musisz napisać funkcję pomocnika, aby jawnie rozpakować argumenty. Jeśli chcesz użyć with, musisz także napisać opakowanie, aby zmienić Poolsię w menedżera kontekstu. (Dzięki mion za wskazanie tego.)

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', ...

W prostszych przypadkach, ze stałym drugim argumentem, możesz także użyć partial, ale tylko w 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. Wiele z tego zostało zainspirowanych jego odpowiedzią, która prawdopodobnie powinna była zostać zaakceptowana. Ale ponieważ ten utknął na górze, najlepszym rozwiązaniem dla przyszłych czytelników wydawało się jego ulepszenie.}

czy istnieje wariant pool.map, który obsługuje wiele argumentów?

Python 3.3 obejmuje pool.starmap()metodę :

#!/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()

W przypadku starszych wersji:

#!/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()

Wynik

1 1
2 1
3 1

Zauważ, jak itertools.izip()i itertools.repeat()są tutaj używane.

Z powodu błędu wspomnianego przez @unutbu nie można używać functools.partial()ani podobnych możliwości w Pythonie 2.6, więc prosta funkcja otoki func_star()powinna być wyraźnie zdefiniowana. Zobacz także obejście sugerowane przezuptimebox .

Myślę, że poniżej będzie lepiej

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

wynik

[3, 5, 7]

Używanie Python 3.3+ zpool.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()

Wynik:

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

Możesz także spakować () więcej argumentów, jeśli chcesz: zip(a,b,c,d,e)

W przypadku, gdy chcemy mieć wartość stałą przekazany jako argument trzeba użyć import itertools, a następnie zip(itertools.repeat(constant), a)na przykład.

Po zapoznaniu się z itertools w odpowiedzi JF Sebastiana postanowiłem pójść o krok dalej i napisać parmappakiet, który dba o równoległość, oferowanie mapi starmapfunkcje w python-2.7 i python-3.2 (i później również), które mogą przyjmować dowolną liczbę argumentów pozycyjnych .

Instalacja

pip install parmap

Jak zrównoleglić:

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)

Przesłałem Parmap do PyPI i do repozytorium github .

Na przykład na pytanie można odpowiedzieć w następujący sposób:

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)

# „Jak wziąć wiele argumentów”.

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)

Istnieje widelec multiprocessingzwany pathos ( uwaga: użyj wersji na github ), który nie potrzebuje starmap- funkcje map odzwierciedlają API dla mapy pythona, więc mapa może przyjmować wiele argumentów. Za pomocą pathosmożna ogólnie wykonywać wiele procesów w interprecie, zamiast utknąć w __main__bloku. Pathos ma się ukazać po niewielkiej aktualizacji - głównie konwersji na Python 3.x.

  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]

pathosma kilka sposobów uzyskania dokładnego zachowania 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]
>>> 

Możesz użyć następujących dwóch funkcji, aby uniknąć pisania opakowania dla każdej nowej funkcji:

import itertools
from multiprocessing import Pool

def universal_worker(input_pair):
    function, args = input_pair
    return function(*args)

def pool_args(function, *args):
    return zip(itertools.repeat(function), zip(*args))

Użyj funkcji functionz listy argumentów arg_0, arg_1a arg_2w następujący sposób:

pool = Pool(n_core)
list_model = pool.map(universal_worker, pool_args(function, arg_0, arg_1, arg_2)
pool.close()
pool.join()

Lepsze rozwiązanie dla python2:

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

2 3 4

1 2 3

0 1 2

na zewnątrz[]:

[3, 5, 7]

Inną prostą alternatywą jest zawinięcie parametrów funkcji w krotkę, a następnie zawinięcie parametrów, które należy przekazać również w krotkach. Być może nie jest to idealne rozwiązanie w przypadku dużych danych. Wierzę, że zrobiłby kopie dla każdej krotki.

from multiprocessing import Pool

def f((a,b,c,d)):
    print a,b,c,d
    return a + b + c +d

if __name__ == '__main__':
    p = Pool(10)
    data = [(i+0,i+1,i+2,i+3) for i in xrange(10)]
    print(p.map(f, data))
    p.close()
    p.join()

Daje wynik w jakiejś losowej kolejności:

0 1 2 3
1 2 3 4
2 3 4 5
3 4 5 6
4 5 6 7
5 6 7 8
7 8 9 10
6 7 8 9
8 9 10 11
9 10 11 12
[6, 10, 14, 18, 22, 26, 30, 34, 38, 42]

Lepszym sposobem jest użycie dekoratora zamiast ręcznego pisania funkcji owijania . Zwłaszcza, gdy masz wiele funkcji do mapowania, dekorator oszczędza Twój czas, unikając pisania opakowań dla każdej funkcji. Zazwyczaj dekorowanej funkcji nie można odebrać, jednak możemy użyć jej functoolsdo obejścia. Więcej dyskusji można znaleźć tutaj .

Oto przykład

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

Następnie możesz zmapować go za pomocą spakowanych argumentów

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

Oczywiście zawsze możesz używać Pool.starmapw Pythonie 3 (> = 3.3), jak wspomniano w innych odpowiedziach.

Innym sposobem jest przekazanie listy list do procedury z jednym argumentem:

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

Można zbudować listę list argumentów za pomocą ulubionej metody.

Oto inny sposób, aby to zrobić, że IMHO jest prostsze i bardziej eleganckie niż jakakolwiek inna podana odpowiedź.

Ten program ma funkcję, która pobiera dwa parametry, drukuje je, a także drukuje sumę:

import multiprocessing

def main():

    with multiprocessing.Pool(10) as pool:
        params = [ (2, 2), (3, 3), (4, 4) ]
        pool.starmap(printSum, params)
    # end with

# end function

def printSum(num1, num2):
    mySum = num1 + num2
    print('num1 = ' + str(num1) + ', num2 = ' + str(num2) + ', sum = ' + str(mySum))
# end function

if __name__ == '__main__':
    main()

wyjście to:

num1 = 2, num2 = 2, sum = 4
num1 = 3, num2 = 3, sum = 6
num1 = 4, num2 = 4, sum = 8

Zobacz dokumentację Pythona, aby uzyskać więcej informacji:

https://docs.python.org/3/library/multiprocessing.html#module-multiprocessing.pool

W szczególności sprawdź starmapfunkcję.

Używam Python 3.6, nie jestem pewien, czy to zadziała ze starszymi wersjami Python

Dlaczego nie ma tak prostego przykładu w dokumentacji, nie jestem pewien.

W Pythonie 3.4.4 można użyć multiprocessing.get_context (), aby uzyskać obiekt kontekstowy do korzystania z wielu metod uruchamiania:

import multiprocessing as mp

def foo(q, h, w):
    q.put(h + ' ' + w)
    print(h + ' ' + w)

if __name__ == '__main__':
    ctx = mp.get_context('spawn')
    q = ctx.Queue()
    p = ctx.Process(target=foo, args=(q,'hello', 'world'))
    p.start()
    print(q.get())
    p.join()

Lub po prostu zastępujesz

pool.map(harvester(text,case),case, 1)

przez:

pool.apply_async(harvester(text,case),case, 1)

Istnieje wiele odpowiedzi tutaj, ale wydaje się, że żadna nie zapewnia kodu zgodnego z Python 2/3, który będzie działał w dowolnej wersji. Jeśli chcesz, aby Twój kod działał , zadziała to dla dowolnej wersji Python:

# For python 2/3 compatibility, define pool context manager
# to support the 'with' statement in Python 2
if sys.version_info[0] == 2:
    from contextlib import contextmanager
    @contextmanager
    def multiprocessing_context(*args, **kwargs):
        pool = multiprocessing.Pool(*args, **kwargs)
        yield pool
        pool.terminate()
else:
    multiprocessing_context = multiprocessing.Pool

Następnie możesz używać wieloprocesowego przetwarzania w zwykły sposób w Pythonie 3, jak chcesz. Na przykład:

def _function_to_run_for_each(x):
       return x.lower()
with multiprocessing_context(processes=3) as pool:
    results = pool.map(_function_to_run_for_each, ['Bob', 'Sue', 'Tim'])    print(results)

będzie działał w Python 2 lub Python 3.

W oficjalnej dokumentacji stwierdzono, że obsługuje tylko jeden iterowalny argument. W takich przypadkach lubię stosować Apply_async. W twoim przypadku zrobiłbym:

from multiprocessing import Process, Pool, Manager

text = "test"
def harvester(text, case, q = None):
 X = case[0]
 res = text+ str(X)
 if q:
  q.put(res)
 return res


def block_until(q, results_queue, until_counter=0):
 i = 0
 while i < until_counter:
  results_queue.put(q.get())
  i+=1

if __name__ == '__main__':
 pool = multiprocessing.Pool(processes=6)
 case = RAW_DATASET
 m = Manager()
 q = m.Queue()
 results_queue = m.Queue() # when it completes results will reside in this queue
 blocking_process = Process(block_until, (q, results_queue, len(case)))
 blocking_process.start()
 for c in case:
  try:
   res = pool.apply_async(harvester, (text, case, q = None))
   res.get(timeout=0.1)
  except:
   pass
 blocking_process.join()

text = "test"

def unpack(args):
    return args[0](*args[1:])

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

if __name__ == '__main__':
    pool = multiprocessing.Pool(processes=6)
    case = RAW_DATASET
    # args is a list of tuples 
    # with the function to execute as the first item in each tuple
    args = [(harvester, text, c) for c in case]
    # doing it this way, we can pass any function
    # and we don't need to define a wrapper for each different function
    # if we need to use more than one
    pool.map(unpack, args)
    pool.close()
    pool.join()

To jest przykład procedury, której używam do przekazywania wielu argumentów do funkcji jednego argumentu używanej w rozwidleniu pool.imap :

from multiprocessing import Pool

# Wrapper of the function to map:
class makefun:
    def __init__(self, var2):
        self.var2 = var2
    def fun(self, i):
        var2 = self.var2
        return var1[i] + var2

# Couple of variables for the example:
var1 = [1, 2, 3, 5, 6, 7, 8]
var2 = [9, 10, 11, 12]

# Open the pool:
pool = Pool(processes=2)

# Wrapper loop
for j in range(len(var2)):
    # Obtain the function to map
    pool_fun = makefun(var2[j]).fun

    # Fork loop
    for i, value in enumerate(pool.imap(pool_fun, range(len(var1))), 0):
        print(var1[i], '+' ,var2[j], '=', value)

# Close the pool
pool.close()

w przypadku python2 możesz użyć tej sztuczki

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

pool = multiprocessing.Pool(processes=6)
b=233
pool.map(lambda x:fun(x,b),range(1000))