使用 dask 的并行化很差答案

【问题标题】：Poor parallelization using dask使用 dask 的并行化很差
【发布时间】：2021-07-20 21:06:08
【问题描述】：

我有一个二维网格，上面有一条路径。我想计算网格的每个点到路径上每个点的距离，然后对这些网格进行一些操作。我正在使用 dask.dataframe 和 dask.array 来完成这项任务。

代码是：

import dask.dataframe as dd
import dask.array as da 

x = np.linspace(-60, 60, 10000)
xv, yv = da.meshgrid(x, x, sparse='True')

path = da.from_array(np.random.rand(100, 2))

h = 100.0

# function to calculate distance to point
def dist_to_point(x, y, p):
    x_dist = x-p[0] 
    y_dist = y-p[1]
    dist = da.sqrt(x_dist**2+y_dist**2)
    d2 = da.sqrt(dist**2 + h**2)    
    return dd.from_dask_array(d2) 


distances = [dist_to_point(xv, yv, path[i, :]) for i in range(npath)]
distances_grid = dd.multi.concat(distances, axis=1, ignore_index=True)

所以distances_grid应该是[到点1的网格距离，到点2的网格距离，...，到点100的网格距离]的串联

现在假设我想在我应用的所有数据帧中获得最大值

l_max = distances_grid.map_partitions(lambda x: x.groupby(level=0, axis=1).max())

这个 dask 图看起来像这样，在我看来这不像是任务的适当并行化。谁能帮我指出我做错了什么或如何改进？我的最终应用程序将在 100000x100000 网格上，因此使用 dask

【问题讨论】：

标签： python numpy dask dask-dataframe

【解决方案1】：

所以万一有人遇到这个问题，我通过广播数组并一起避免 for 循环来解决它。我最终使用的代码是

x = da.from_array(np.linspace(-60, 60, 10000), chunks=1000)
xv, yv = da.meshgrid(x, x, sparse='True')
path = da.from_array(np.random.rand(10, 2))

h = 100.0

ngrid = x.shape[0]

xd = x[:, np.newaxis] - path[:, 0]
yd = x[:, np.newaxis] - path[:, 1]
z = xd**2 + yd[:, np.newaxis]**2 + h**2

# euclidian distance at height = 100
z = xd**2 + yd[:, np.newaxis]**2 + h**2
distances_grid = z**0.5

l_max = distances_grid.max(axis=2)

这给了我一个更好的图表，我可以通过改变块的大小来更好地平衡它。

【讨论】：