Paralléliser une boucle for imbriquée en IPython

Question

J'ai une boucle for imbriquée dans mon code python qui ressemble à ceci :

results = []

for azimuth in azimuths:
    for zenith in zeniths:
        # Do various bits of stuff
        # Eventually get a result
        results.append(result)

J'aimerais paralléliser cette boucle sur ma machine à 4 cœurs pour l'accélérer. En regardant la documentation sur la programmation parallèle d'IPython (http://ipython.org/ipython-doc/dev/parallel/parallel\_multiengine.html#quick-and-easy-parallelism), il semble qu'il y ait un moyen facile d'utiliser map pour paralléliser les opérations itératives.

Cependant, pour ce faire, je dois avoir le code à l'intérieur de la boucle comme une fonction (ce qui est facile à faire), et ensuite mapper à travers cette fonction. Le problème que je rencontre est que je n'arrive pas à obtenir un tableau pour mapper cette fonction. itertools.product() produit un itérateur avec lequel je n'arrive pas à utiliser la fonction map.

Est-ce que je fais fausse route en essayant d'utiliser map ici ? Existe-t-il une meilleure façon de procéder ? Ou existe-t-il un moyen d'utiliser itertools.product et de procéder ensuite à une exécution parallèle à l'aide d'une fonction mappée sur les résultats ?

Answer 1

Pour paralléliser chaque appel, il suffit d'obtenir une liste pour chaque argument. Vous pouvez utiliser itertools.product + zip pour obtenir ceci :

allzeniths, allazimuths = zip(*itertools.product(zeniths, azimuths))

Vous pouvez alors utiliser la carte :

amr = dview.map(f, allzeniths, allazimuths)

Pour approfondir les étapes, voici un exemple :

zeniths = range(1,4)
azimuths = range(6,8)

product = list(itertools.product(zeniths, azimuths))
# [(1, 6), (1, 7), (2, 6), (2, 7), (3, 6), (3, 7)]

Nous avons donc une "liste de paires", mais ce que nous voulons vraiment, c'est une liste unique pour chaque argument, c'est-à-dire une "paire de listes". C'est exactement ce que la fonction un peu bizarre de zip(*product) la syntaxe nous atteint :

allzeniths, allazimuths = zip(*itertools.product(zeniths, azimuths))

print allzeniths
# (1, 1, 2, 2, 3, 3)
print allazimuths
# (6, 7, 6, 7, 6, 7)

Il ne nous reste plus qu'à mapper notre fonction sur ces deux listes, afin de paralléliser les boucles for imbriquées :

def f(z,a):
    return z*a

view.map(f, allzeniths, allazimuths)

Et il n'y a rien de spécial à ce qu'il n'y en ait que deux - cette méthode devrait s'étendre à un nombre arbitraire de boucles imbriquées.

Answer 2

Je suppose que vous utilisez IPython 0.11 ou une version ultérieure. Tout d'abord, définissez une fonction simple.

def foo(azimuth, zenith):
    # Do various bits of stuff
    # Eventually get a result
    return result

utilisez ensuite la suite parallèle d'IPython pour paralléliser votre problème. démarrez d'abord un contrôleur avec 5 moteurs attachés (#CPUs + 1) en démarrant un cluster dans une fenêtre de terminal (si vous avez installé IPython 0.11 ou une version ultérieure, ce programme devrait être présent) :

ipcluster start -n 5

Dans votre script, connectez-vous au contrôleur et transmettez toutes vos tâches. Le contrôleur s'occupera de tout.

from IPython.parallel import Client

c = Client()   # here is where the client establishes the connection
lv = c.load_balanced_view()   # this object represents the engines (workers)

tasks = []
for azimuth in azimuths:
    for zenith in zeniths:
        tasks.append(lv.apply(foo, azimuth, zenith))

result = [task.get() for task in tasks]  # blocks until all results are back

Answer 3

Je ne suis pas vraiment familier avec IPython, mais une solution facile semblerait être de paralléliser uniquement la boucle extérieure.

def f(azimuth):
    results = []
    for zenith in zeniths:
        #compute result
        results.append(result)
    return results

allresults = map(f, azimuths)

Answer 4

Si vous souhaitez réellement exécuter votre code en parallèle, utilisez concurrent.futures

import itertools
import concurrent.futures

def _work_horse(azimuth, zenith):
    #DO HEAVY WORK HERE
    return result

futures = []
with concurrent.futures.ProcessPoolExecutor() as executor:
    for arg_set in itertools.product(zeniths, azimuths):
        futures.append(executor.submit(_work_horse, *arg_set))
executor.shutdown(wait=True)

# Will time out after one hour.
results = [future.result(3600) for future in futures]

Answer 5

Si vous souhaitez conserver la structure de votre boucle, vous pouvez essayer d'utiliser Ray ( documents ), qui est un cadre pour l'écriture de Python parallèle et distribué. La seule exigence est que vous devez séparer le travail qui peut être parallélisé dans sa propre fonction.

Vous pouvez importer Ray comme suit :

import ray

# Start Ray. This creates some processes that can do work in parallel.
ray.init()

Votre script ressemblerait alors à ceci :

# Add this line to signify that the function can be run in parallel (as a
# "task"). Ray will load-balance different `work` tasks automatically.
@ray.remote
def work(azimuth, zenith):
  # Do various bits of stuff
  # Eventually get a result
  return result

results = []

for azimuth in azimuths:
    for zenith in zeniths:
        # Store a future, which represents the future result of `work`.
        results.append(work.remote(azimuth, zenith))

# Block until the results are ready with `ray.get`.
results = ray.get(results)