Dask任务因内存占满被终止？如何限制其内存使用？

首先明确：内存占满后任务被终止不是Dask的预期行为——正常配置下，当内存接近阈值时，Dask应该会把部分数据溢写到磁盘，避免触发系统的OOM（内存不足）杀手。你遇到的情况大概率是内存限制的配置没生效，或者操作本身的内存需求没被正确管控。

为什么dask.set_options(available_memory=12e9)没生效？ #

这个写法已经过时啦！在新版本的Dask中，available_memory参数已经被废弃，不再起作用。现在统一使用memory_limit来配置内存限制，而且配置方式要看你用的是单机模式还是分布式模式：

正确的内存限制配置方法 #

• 单机模式（无分布式调度器）：
  用dask.config.set来全局设置内存上限：

  import dask
  dask.config.set({"memory_limit": "12GB"})  # 可以用字符串或者数值（比如12e9）
  

  或者在读取数据时就控制分块大小，避免单个分块占用过多内存：

  df = dask.dataframe.read_csv("large_file.csv", blocksize="256MB")  # 根据你的内存调整块大小
  

• 分布式模式（推荐用于大内存任务）：
  启动本地集群时直接指定每个worker的内存限制，这样调度器能更精准地管控内存：

  from dask.distributed import Client
  # 每个worker分配12GB内存，如果你有多个worker可以调整n_workers参数
  client = Client(memory_limit="12GB", n_workers=1)
  

为什么数据没自动溢写到磁盘？ #

除了配置问题，还有几个可能的原因：

• 单个分块过大：如果你的DataFrame分块尺寸超过了内存限制，Dask没办法把单个分块拆成更小的部分溢写，这时候就会直接占满内存。解决办法是在读取数据时设置更小的blocksize，或者用repartition手动拆分分块。
• shuffle类操作的内存压力：像groupby、merge、sort_values这类需要shuffle数据的操作，默认可能会优先用内存处理。你可以强制开启磁盘shuffle来减少内存占用：

  df.groupby("column").sum(shuffle_method="disk")
  

• 临时目录空间问题：虽然你说磁盘剩余充足，但要确认Dask使用的临时目录（默认是系统的/tmp）有没有足够空间。可以手动指定临时目录：

  dask.config.set({"temporary_directory": "/path/to/your/large/disk/dir"})
  

额外的调试建议 #

• 如果你用了分布式调度器，打开client.dashboard（默认地址是http://localhost:8787），可以实时查看每个worker的内存使用情况，定位是哪个步骤占用了过多内存。
• 用df.memory_usage().compute()查看单个分块的内存占用，确保它远小于你的内存限制。

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