ipython map_async 输入输出数据

Question

我是 IPython 并行包的新手，但我真的很想继续使用它。我拥有的是一个 4D numpy 数组，我想通过切片、行、列运行并处理第 4 维（时间）。该处理是一个最小化例程，需要一些时间，这就是我想并行化它的原因。

from IPython.parallel import Client
from numpy import *
from matplotlib.pylab import *

c = Client()

v = c.load_balanced_view()
v.block=False

def process( src, freq, d ):
        # Get slice, row, col
        sl,r,c = src

        # Get data
        mm = d[:,sl,c,r]

        # Call fitting routine
        <fiting routine that requires freq, mm and outputs multiple parameters> 

        return <output parameters??>


##  Create the mask of what we are going to process
mask = zeros(d[0].shape)
mask[sl][ nonzero( d[0,sl] > 10*median(d[0]) ) ] = 1

# find all non-zero points in the mask
points = array(nonzero( mask == 1)).transpose()

# Call async
asyncresult = v.map_async( process, points, freq=freq, d=d )

我的函数“process”需要两个参数：1) freq 是一个 numpy 数组 (100,1) 和 2) d 是 (100, 50, 110, 110) 左右。我想从配件中检索几个参数。

我看到的所有使用 map_async 的示例都有简单的 lambda 函数等，而且输出似乎很简单。

我想要的是对 d 中掩码不为零的每个点应用“处理”，并将输出参数的映射放在同一空间中。 [补充：我得到“process() 正好需要 3 个参数（1 个给定）]。

（这可能需要第 2 步，因为我将一个巨大的 numpy 数组“d”传递给每个进程。但是一旦我弄清楚数据传递，我应该希望能够找到一种更有效的方法来执行此操作。 )

感谢您的帮助。

Answer 1

我通过这样做解决了数据传递问题

def mapper(x):
    return apply(x[0], x[1:])

并使用元组列表调用 map_async，其中第一个元素是我的函数，其余元素是我的函数的参数。

asyncResult = pool.map_async(mapper, [(func, arg1, arg2) for arg1, arg2 in myArgs])

我先尝试了一个 lambda，但显然它不能被腌制，所以这是不行的。