使用Dask delayed并行化自定义函数性能劣于串行的问题排查

你的并行版本性能劣于串行的核心原因有两个：

• CPython的GIL限制：Dask默认使用线程池调度器（threads），但纯Python编写的CPU密集型循环会被全局解释器锁（GIL）限制——多个线程无法真正同时执行CPU任务，反而要额外承担线程切换、任务调度的开销，导致整体速度下降。
• 冗余的调度逻辑：代码里用dask.delayed(ss)包装整个任务列表属于多余操作，会增加不必要的调度成本。

修复方案 #

1. 切换到进程池调度器 #

对于CPU密集型任务，必须用进程池绕过GIL，在compute()时指定scheduler='processes'即可。

2. 对齐计算范围并简化调度 #

原并行代码的计算范围和串行版本不一致（串行从1开始，并行第一个任务从0开始），需要调整区间保证结果准确；同时直接对延迟任务列表调用dask.compute(*ss)，无需额外包装。

修改后的代码 #

import dask, time
import os

def mysum(rng):
    z = 0
    for i in rng:
        z += i
    return z

# 串行版本
b = time.time()
zz = mysum(range(1, 1_000_000_000))
t = time.time() - b
print(f'time to run in serial {t}')

# 并行版本
ms_parallel = dask.delayed(mysum)
ss = []
# 建议使用实际核心数，避免过度调度
ncores = os.cpu_count()
m = 100_000_000
for i in range(ncores):
    lower = m * i + 1
    # 最后一个任务的上限对齐串行版本的终点
    upper = (i + 1) * m + 1 if i != ncores-1 else 1_000_000_000
    r = range(lower, upper)
    s = ms_parallel(r)
    ss.append(s)

b = time.time()
# 使用进程池调度器实现真正并行
yy = dask.compute(*ss, scheduler='processes')
t = time.time() - b
print(f'time to run in parallel {t}')
# 验证并行与串行结果一致
assert sum(yy) == zz

额外说明 #

进程池的缺点是进程间数据传输存在额外开销，但对于CPU密集型任务，这个开销远小于并行带来的计算收益；如果你的机器核心数较少，可适当调整ncores数值，避免进程切换过于频繁。

内容的提问来源于stack exchange，提问作者SednaGammaPrime