如何在dask中并行化groupby（）？

Question

我试过了：

df.groupby('name').agg('count').compute(num_workers=1)
df.groupby('name').agg('count').compute(num_workers=4)

他们花相同的时间，为什么num_workers不起作用？

谢谢

Answer 1

默认情况下，Dask 将处理多线程任务，这意味着它在您的计算机上使用单个处理器。（请注意，如果您的数据无法放入内存，则使用 dask 仍然很有趣）

如果要使用多个处理器来计算操作，则必须使用不同的调度程序：

from dask import dataframe as dd
from dask.distributed import LocalCluster, Client

df = dd.read_csv("data.csv")

def group(num_workers): 
    start = time.time() 
    res = df.groupby("name").agg("count").compute(num_workers=num_workers) 
    end = time.time() 
    return res, end-start

print(group(4))

clust = LocalCluster()
clt = Client(clust, set_as_default=True) 
print(group(4)) 

在这里，我使用 4 个并行进程（因为我有一个四核）创建了一个本地集群，然后设置了一个默认调度客户端，它将使用这个本地集群来执行 Dask 操作。使用 1.5 Gb 的 CSV 两列文件，标准 groupby 在我的笔记本电脑上需要大约 35 秒，而多进程只需要大约 22 秒。