如何让Dask延迟对象内存超限时返回默认值及优化内存占用？

我需要并行评估大数据集上的机器学习管道列表，采用循环生成模型/管道后评估的方式，当前并行执行代码如下：

for i in range(10):
   pipeline_list = generate_next_pipelines()
   scores = dask.compute(*[dask.delayed(fit_and_score)(pipeline, X, y) for pipeline in pipeline_list])
   # save/print scores

目前遇到未管理内存占用过高的错误，想了解是否有遗漏的步骤可减少内存占用或更频繁释放未释放内存？

我已通过设置LocalCluster将内存限制设为系统最大值，代码可运行但任务请求内存超出时整个脚本崩溃，希望Dask在指定工作节点内存超限时返回“内存不足”这类默认值。相关集群配置代码如下：

cluster = LocalCluster(n_workers=n_jobs, 
                       threads_per_worker=1,
                       memory_limit='64GB')
client = Client(cluster)

一、主动清理内存与任务缓存 #

每次循环结束后显式清理任务图、结果对象及工作节点缓存，避免内存累积：

for i in range(10):
   pipeline_list = generate_next_pipelines()
   delayed_tasks = [dask.delayed(fit_and_score)(pipeline, X, y) for pipeline in pipeline_list]
   scores = dask.compute(*delayed_tasks)
   # save/print scores
   
   # 清理本地对象
   del delayed_tasks, scores
   import gc
   gc.collect()
   # 清理Dask集群缓存与未完成任务
   client.cancel(client.get_tasks())
   client.run(gc.collect)  # 触发工作节点垃圾回收

二、配置集群内存阈值，避免直接崩溃 #

调整LocalCluster的内存监控参数，让集群在内存接近阈值时逐步预警、溢写磁盘，而非直接终止进程：

cluster = LocalCluster(
    n_workers=n_jobs,
    threads_per_worker=1,
    memory_limit='64GB',
    memory_target_fraction=0.6,  # 内存使用达60%时触发预警
    memory_spill_fraction=0.7,   # 达70%时将数据溢写到磁盘
    memory_pause_fraction=0.8,   # 达80%时暂停新任务调度
    memory_terminate_fraction=0.9  # 达90%时终止超内存任务（可按需调整）
)
client = Client(cluster)

三、让任务返回内存不足默认值 #

在fit_and_score函数中捕获内存异常，返回自定义默认值：

def fit_and_score(pipeline, X, y):
    try:
        pipeline.fit(X, y)
        return pipeline.score(X, y)
    except MemoryError:
        return "内存不足"
    except Exception as e:
        print(f"任务执行失败: {str(e)}")
        return "任务失败"

同时可给延迟任务添加重试机制，避免单次内存波动导致失败：

delayed_tasks = [dask.delayed(fit_and_score, retries=1)(pipeline, X, y) for pipeline in pipeline_list]

四、其他内存优化技巧 #

• 分块处理数据集：如果X、y是大型数据集，改用Dask DataFrame/Array分块存储，让单任务仅处理部分数据，降低内存压力。
• 分批执行任务：不要一次性提交所有管道任务，用as_completed分批处理，控制并行任务数：

from dask.distributed import as_completed

for i in range(10):
   pipeline_list = generate_next_pipelines()
   futures = client.map(fit_and_score, pipeline_list, X=X, y=y)
   # 逐个获取结果，避免一次性加载所有结果到内存
   for future in as_completed(futures):
       score = future.result()
       # save/print score
   del futures
   gc.collect()

• 轻量化管道传递：若管道对象过大，可只传递管道配置参数，在任务内部重建管道，减少数据传输与内存占用。

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