如何让Dask使用输入引用而非创建输入副本？

你的问题核心是错误地用dask.delayed包裹了Dask数组操作，导致Dask无法优化任务依赖，进而产生不必要的重复计算或临时内存占用，并非创建了600份X的副本。以下是具体的解决方案：

方案1：直接使用Dask数组原生操作（推荐） #

Dask数组本身就支持延迟计算和任务优化，完全不需要用dask.delayed来包裹简单的聚合操作。把你的代码改成这样：

import numpy as np
import dask.array as da

def foo(a):
    return a.sum()

x = np.random.rand(1000000,70)
X = da.from_array(x)

# 直接生成Dask数组的sum对象列表，而非delayed对象
sums_list = [X.sum() for _ in range(600)]
Xsums = da.compute(*sums_list)

这样Dask会自动识别所有sum操作共享同一个X的依赖，只会计算一次X的分区数据，然后复用结果完成所有600次求和，内存占用会大幅降低。

方案2：若必须使用dask.delayed，先持久化X #

如果你的foo函数包含更复杂的自定义逻辑（无法用Dask数组原生操作实现），可以先将X持久化到内存，确保所有delayed任务复用同一个内存中的数据：

import numpy as np
import dask
import dask.array as da

def foo(a):
    return a.sum()

x = np.random.rand(1000000,70)
X = da.from_array(x)

# 持久化X到内存，避免重复加载
X_persisted = dask.persist(X)[0]

X_list = [dask.delayed(foo)(X_persisted) for n in range(600)]
Xsums = dask.compute(*X_list)

dask.persist会将X的分区加载到内存并保留，所有后续任务都会直接引用这个内存中的数据，不会产生副本。

为什么原代码会内存溢出？ #

原代码中，dask.delayed(foo)(X)会将X作为普通输入序列化，Dask对delayed任务的依赖优化能力远弱于原生Dask数组操作。这会导致调度器可能重复计算X的分区，或者在内存中保留多个临时计算结果，最终引发内存占用过高。而原生Dask数组操作会自动合并共享依赖，最大化复用中间结果。

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