IPython 中带多个参数的并行映射函数

我正在尝试使用 IPython 的并行环境，到目前为止，它看起来很棒，但我遇到了问题。假设我有一个在库中定义的函数

def func(a,b):
   ...

当我想要评估 a 的一个值和 b 的一组值时，我会使用它。

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

显然，真正的函数更复杂，但问题的本质是它需要多个参数，而我只想映射其中一个参数。问题是 map、@dview.parallel 等映射了所有参数。

假设我想得到 func(myA, myLongList) 的答案。最明显的方法是使用 functools.partial 进行柯里化，或者就像

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

但是，这在远程计算机上无法正常工作。原因是，当 lambda 表达式被 pickle 时，不包括 myA 的值，而是使用远程计算机上本地范围内的 myA 值。当闭包被腌制时，它们关闭的变量却不会。

我能想到的两种实际可行的方法是手动为每个参数构建列表并对所有参数进行映射，

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

或者可怕地使用数据作为函数的默认参数，迫使它被腌制：

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

确实，当实际函数需要大量参数并且更加复杂时，这一切看起来都非常扭曲。有一些惯用的方法吗？就像是

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

但据我所知，像“mapOver”这样的东西并不存在。我可能知道如何实现它......这感觉像是一个非常基本的操作，应该存在支持，所以我想检查我是否遗漏了一些东西。

我可以对巴图的答案进行一些改进（我认为这是一个很好的答案，但可能没有详细说明为什么使用这些选项）。 ipython 文档目前在这一点上也严重不足。所以你的函数的形式是：

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

并存储在一个名为mylib。假设 b、c 和 d 在运行期间相同，因此您编写一个 lambda 函数将其简化为 1 参数函数。

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

并且你想运行：

z=dview.map_sync(mylamfxn, iterable_of_a)

在梦想的世界里，一切都会像那样神奇地进行。但是，首先您会收到“未找到 mylib”的错误，因为 ipcluster 进程尚未加载 mylib。如有必要，请确保 ipcluster 进程的 python 路径中有“mylib”，并且位于 myfxn 的正确工作目录中。然后你需要添加到你的Python代码中：

dview.execute('import mylib')

它运行import mylib每个进程的命令。如果再试一次，您将收到类似“全局变量 b 未定义”的错误，因为虽然变量位于 Python 会话中，但它们并不位于 ipcluster 进程中。然而，python提供了一种将一组变量复制到子进程的方法。继续上面的例子：

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

现在所有子进程都可以访问 b、c 和 d。然后你可以运行：

z=dview.map_sync(mylamfxn, iterable_of_a)

现在它应该像广告中那样工作了。不管怎样，我是Python并行计算的新手，并且发现这个线程很有用，所以我想我应该尝试帮助解释一些让我有点困惑的点......

最终的代码是：

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)

这可能是让几乎任何令人尴尬的并行代码在 python 中并行运行的最快、最简单的方法......

UPDATE您还可以使用 dview 不循环地推送所有数据，然后使用 lview （即lview=rc.load_balanced_view(); lview.map(...)以负载平衡的方式进行实际计算。