使用Dask from_delayed()加载DataFrame时内存占用过高问题求助

关于dd.from_delayed的扩展性 #

dd.from_delayed本身具备良好扩展性，但你当前的使用方式触发了客户端内存瓶颈：当一次性创建10000个delayed对象并传给from_delayed时，Dask需要在客户端一次性构建所有任务的依赖图，万级任务的元数据（包括任务定义、DataFrame结构信息等）会集中占用客户端内存，导致内存耗尽进而影响Worker调度。而Dask Array的任务划分逻辑更紧凑，元数据占用更低，因此未出现相同问题。

替代的分批加载方式 #

1. 分批次创建Dask DataFrame再合并 #

将文件名拆分为若干小批次（比如每100个一批），每批生成一个小型Dask DataFrame，最后用dd.concat合并，分散客户端的元数据处理压力：

import dask.dataframe as dd
from dask.delayed import delayed

batch_size = 100  # 根据内存情况调整批次大小
partial_dfs = []

for idx in range(0, len(filenames), batch_size):
    batch_files = filenames[idx:idx+batch_size]
    batch_delayed = [delayed(load)(fn) for fn in batch_files]
    partial_df = dd.from_delayed(batch_delayed, meta=types)
    partial_dfs.append(partial_df)

final_df = dd.concat(partial_dfs, axis=0)

2. 用Dask Bag中转后转DataFrame #

Dask Bag天生适合处理大量小文件的分批加载，元数据占用远低于直接创建万级delayed对象，之后再转换为Dask DataFrame：

import dask.bag as db
import dask.dataframe as dd

# 从文件名序列创建Bag，映射加载函数
file_bag = db.from_sequence(filenames).map(load)
# 转换为DataFrame，指定元数据
final_df = file_bag.to_dataframe(meta=types)

3. 优先使用Dask内置读取函数（若适用） #

如果你的load函数只是封装了Pandas的read_csv/read_parquet等常规读取逻辑，直接使用Dask DataFrame的内置读取函数是最优解——这些函数内部已做了任务分区、元数据优化，能自动避免客户端内存过载：

# 示例：CSV文件
final_df = dd.read_csv(filenames, dtype=types)
# 示例：Parquet文件
final_df = dd.read_parquet(filenames, schema=types)

4. 调整任务图优化参数（辅助方案） #

通过Dask配置减少任务图的内存占用，比如禁用纯函数缓存（仅在必要时使用）：

import dask

dask.config.set({"delayed_pure": False})

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