在单台机器上并行 dask 代码的问题答案

【问题标题】：problem parralleling dask code on single machine在单台机器上并行 dask 代码的问题
【发布时间】：2019-04-22 12:11:46
【问题描述】：

与 dask 并行比顺序编码慢。

我有一个嵌套的 for 循环，我试图在本地集群上并行，但找不到正确的方法。

我想并行内部循环。

我有 2 个大的 numpy 矩阵，我试图对其进行迭代并对矩阵的子集执行数学计算。尺寸：

data_mat.shape = (38, 243863)
indicies_mat.shape (243863, 27)
idxX.shape = (19,)
idxY.shape = (19,)

seq_code：

start = datetime.datetime.now()
for i in range(num+1):
    if i == 0:
        labels = np.array(true_labels)
    else:
        labels = label_mat[i]

    idxX = list(np.where(labels == 1))
    idxY = list(np.where(labels == 2))

    ansColumn = []

    for j in range(indices.shape[0]):
        list_of_indices = [[i] for i in indices_slice]
        dataX = (data_mat[idxX, list_of_indices]).T
        dataY = (data_mat[idxY, list_of_indices]).T
        ansColumn.append(calc_func(dataX, dataY))

    if i == 0:
        ansMat = ansColumn
    else:
        ansMat = np.c_[ansMat, ansColumn]


end = datetime.datetime.now()
print(end - start)

并行代码：

start = datetime.datetime.now()
cluster = LocalCluster(n_workers=4, processes=False)
client = Client(cluster)
for i in range(num+1):
    if i == 0:
        labels = np.array(true_labels)
    else:
        labels = label_mat[i]

    idxX = list(np.where(labels == 1))
    idxY = list(np.where(labels == 2))

    [big_future] = client.scatter([data_mat], broadcast=True)
    [idx_b] = client.scatter([idxX], broadcast=True)
    [idy_b] = client.scatter([idxY], broadcast=True)


    futures = [client.submit(prep_calc_func, idx_b, idy_b, indices[j, :], big_future) for j in range(indices.shape[0])]
    ansColumn = []

    for fut in dask.distributed.client.as_completed(futures):
        ansColumn.append(fut.result())

    if i == 0:
        ansMat = ansColumn
    else:
        ansMat = np.c_[ansMat, ansColumn]


end = datetime.datetime.now()
print(end - start)

辅助函数：

def = prep_calc_func(idxX, idxY, subset_of_indices, data_mat):
    list_of_indices = [[i] for i in indices_slice]
    dataX = (data_mat[idxX, subset_of_indices]).T
    dataY = (data_mat[idxY, subset_of_indices]).T
    ret_val = calc_func(dataX, dataY)
    return ret_val

本地机器：MacBook Pro（Retina，13 英寸，2014 年中）处理器：2.6 GHz Intel Core i5

hw.physicalcpu: 2 hw.logicalcpu: 4

内存：8 GB 1600 MHz DDR3

当我执行 seq 代码时，需要 01:52 分钟才能完成（不到 2 分钟）

但是当我尝试并行代码时，它需要超过 15 分钟。（无论我使用哪种方法：计算、结果和 client.submit 或 dask 延迟）

（我更喜欢使用 dask 分布式包，因为下一阶段可能也会使用远程集群。）

知道我做错了什么吗？

【问题讨论】：

标签： python numpy dask dask-distributed

【解决方案1】：

有很多原因会导致速度变慢。可能会有很多交流。你的任务可能太小了（回想一下 Dask 的开销大约是每个任务 1 毫秒），或者完全是其他的东西。有关了解 Dask 性能的更多信息，我推荐以下文档：

【讨论】：