Fonctions de mappage parallèle dans IPython avec plusieurs paramètres

Question

Je suis en train d'utiliser IPython parallèles de l'environnement et pour l'instant, il a l'air super mais je suis confronté à un problème. Disons que j'ai une fonction, définie dans une bibliothèque

def func(a,b):
   ...

que j'utilise quand je veux évaluer sur une valeur de un et un tas de valeurs de b.

[func(myA, b) for b in myLongList]

De toute évidence, la fonction réelle est plus compliquée, mais l'essence de la matière est qu'elle prend plusieurs paramètres et j'aimerais carte sur un seul d'entre eux. Le problème est que la carte, @dview.en parallèle, etc. carte plus de tous les arguments.

Donc disons que je veux obtenir la réponse à la touche func(myA, myLongList). Le moyen le plus évident pour ce faire est de curry, soit w/ functools.partielle ou juste comme

dview.map_sync(lambda b: func(myA, b),   myLongList)

Toutefois, cela ne fonctionne pas correctement sur des machines distantes. La raison en est que, lorsque l'expression lambda est décapé, la valeur de myA n'est pas compris et au lieu de cela, la valeur de myA de l'étendue locale sur l'ordinateur distant est utilisé. Lors de fermetures d'obtenir marinés, les variables de fermer plus de ne pas.

Deux façons que je peux penser de ce qui sera effectivement le travail sont à construire manuellement des listes pour chaque argument et avoir une carte de travail de plus de tous les arguments,

dview.map_sync(func, [myA]*len(myLongList), myLongList)   

ou de terrifiant utiliser les données par défaut des arguments à une fonction, en l'obligeant à obtenir marinés:

# Can't use a lambda here b/c lambdas don't use default arguments :(
def parallelFunc(b, myA = myA):
    return func(myA, b)

dview.map_sync(parallelFunc, myLongList)

Vraiment, tout cela semble horriblement tordu lorsque la fonction réelle prend beaucoup de paramètres, et plus compliqué. Est-il un idiomatiques façon de faire cela? Quelque chose comme

@parallel(mapOver='b')
def  bigLongFn(a, b):
   ...

mais autant que je sache, rien de tel que le " mapOver chose existe. J'ai probablement une idée de comment le mettre en œuvre ... cela se sent juste comme un très fonctionnement de base qu'il doit exister un soutien pour si je veux vérifier si je suis en manque de quelque chose.

Answer 1

Je peux améliorer un peu sur batu réponse (qui je pense est une bonne idée, mais n'a pas peut-être le document le plus de détails POURQUOI vous utilisez ces options). Le ipython documentation est également actuellement lamentablement sur ce point. Si votre fonction est de la forme:

def myfxn(a,b,c,d):
  ....
  return z

et stockées dans un fichier appelé mylib. Disons b,c, et d sont les mêmes au cours de votre course, de sorte que vous écrivez une fonction lambda pour le réduire à un 1-paramètre de la fonction.

import mylib
mylamfxn=lambda a:mylib.myfxn(a,b,c,d)

et vous souhaitez exécuter:

z=dview.map_sync(mylamfxn, iterable_of_a)

Dans un monde de rêve, tout comme par magie comme ça. Cependant, tout d'abord, vous obtiendrez une erreur de "mylib pas trouvé," parce que le ipcluster processus n'avez pas chargé mylib. Assurez-vous que le ipcluster processus de "mylib" dans leurs python chemin et sont dans le bon répertoire de travail pour myfxn, si nécessaire. Ensuite, vous devez ajouter à votre code python:

dview.execute('import mylib')

qui exécute l' import mylib commande sur chaque processus. Si vous essayez à nouveau, vous recevrez un message d'erreur le long des lignes de "variable globale b non définie", car tandis que les variables sont dans votre python session, ils ne sont pas dans le ipcluster processus. Cependant, python fournit une méthode de copie d'un groupe de variables à la sous-processus. En reprenant l'exemple ci-dessus:

mydict=dict(b=b, c=c, d=d)
dview.push(mydict)

Maintenant, tous les sous-processus ont accès aux b,c et d. Ensuite, il vous suffit d'exécuter:

z=dview.map_sync(mylamfxn, iterable_of_a)

et il devrait maintenant fonctionner comme annoncé. De toute façon, je suis nouveau sur le calcul parallèle avec python, et j'ai trouvé ce thread utile, donc je pensais que je voudrais essayer pour aider à expliquer quelques-uns des points qui me troublait un peu....

La version finale du code serait:

import mylib

#set up parallel processes, start ipcluster from command line prior!
from IPython.parallel import Client
rc=Client()
dview=rc[:]

#...do stuff to get iterable_of_a and b,c,d....

mylamfxn=lambda a:mylib.myfxn(a,b,c,d)

dview.execute('import mylib')
mydict=dict(b=b, c=c, d=d)
dview.push(mydict)
z=dview.map_sync(mylamfxn, iterable_of_a)

C'est probablement le moyen le plus rapide et le plus simple de faire pratiquement toute embarrassant de code parallèle parallèle en python....

Mise à JOUR que Vous pouvez également utiliser dview à pousser toutes les données sans boucles et ensuite utiliser un lview (c - lview=rc.load_balanced_view(); lview.map(...) pour faire le calcul de la charge de façon équilibrée.

Answer 2

Ceci est mon premier message à StackOverflow, de sorte s'il vous plaît être doux ;) j'ai essayé de faire la même chose, et est venu avec ce qui suit. Je suis assez sûr que ce n'est pas le moyen le plus efficace, mais semble fonctionner un peu. Une mise en garde pour l'instant, c'est que pour une raison que je ne vois que deux moteurs de travailler à 100%, les autres sont assis quasiment inactif...

Pour faire un appel à un multiple de l'argument de la fonction de la carte j'ai d'abord écrit cette routine dans mon parallel.py module:

def map(r,func, args=None, modules=None):
"""
Before you run parallel.map, start your cluster (e.g. ipcluster start -n 4)

map(r,func, args=None, modules=None):
args=dict(arg0=arg0,...)
modules='numpy, scipy'    

examples:
func= lambda x: numpy.random.rand()**2.
z=parallel.map(r_[0:1000], func, modules='numpy, numpy.random')
plot(z)

A=ones((1000,1000));
l=range(0,1000)
func=lambda x : A[x,l]**2.
z=parallel.map(r_[0:1000], func, dict(A=A, l=l))
z=array(z)

"""
from IPython.parallel import Client
mec = Client()
mec.clear()
lview=mec.load_balanced_view()
for k in mec.ids:
  mec[k].activate()
  if args is not None:
    mec[k].push(args)
  if modules is not None:
    mec[k].execute('import '+modules)
z=lview.map(func, r)
out=z.get()
return out

Comme vous pouvez le voir, la fonction prend un paramètre args qui est un dict de paramètres dans la tête des nœuds de l'espace de travail. Ces paramètres sont ensuite transmis aux moteurs. À ce point qu'ils deviennent des objets locaux et peut être utilisé dans la fonction directement. Par exemple, dans le dernier exemple donné ci-dessus dans les commentaires, l'Une matrice est découpé à l'aide de la l moteur variable locale.

Je dois dire que même si la fonction ci-dessus fonctionne, je ne suis pas 100% heureux avec elle pour le moment. Si je peux trouver quelque chose de mieux, je vais le poster ici.

Mise à JOUR:2013/04/11 J'ai apporté des modifications mineures au code: - Activer l'instruction est manquante entre crochets. L'amenant à ne pas courir. - Déplacement du mec.clear() en haut de la fonction, par opposition à la fin. J'ai aussi remarqué que cela fonctionne mieux si je le lance dans ipython. Par exemple, j'pouvez obtenir des erreurs si j'exécute un script à l'aide de la fonction ci-dessus comme "python ./myparallelrun.py" mais pas si je le lance dans ipython à l'aide de "%exécutez ./myparallelrun.py". Je ne sais pas pourquoi...

Answer 3

Une façon élégante de le faire est d'utiliser des fonctions partielles.

Si vous savez que vous voulez que le premier argument de foo soit myArg, vous pouvez créer une nouvelle barre de fonction en

 from functools import partial
bar = partial(foo, myARg)
 

bar(otherArg) retournera alors foo(myArg,otherArg)

Answer 4

bâtissons sur cela:

 dview.map_sync(func, [myA]*len(myLongList), myLongList)
 

peut-être que les choses suivantes fonctionneraient:

 from itertools import izip_longest
dview.map_sync(func, izip_longest(myLongList, [], fillvalue=myA))
 

Exemple:

 >>> # notice that a is a tuple
... concat = lambda a: '%s %s' % a
>>> mylonglist = range(10)
>>> from itertools import izip_longest
>>> map(concat, izip_longest(mylonglist, [], fillvalue='mississippi'))
['0 mississippi', '1 mississippi', '2 mississippi', '3 mississippi',
'4 mississippi', '5 mississippi', '6 mississippi', '7 mississippi',
'8 mississippi', '9 mississippi']