Dask本地集群加载数据提示Worker内存超限重启、KilledWorker错误如何解决？

• 校验JSON文件大小：确认2019-12-16-latest-level.json单文件大小，排查是否超过单worker分配的内存（当前集群单worker内存约31.45GiB）
• 检查默认分块合理性：db.read_text默认分块大小为64MB，如果行分隔JSON文件存在单行长超过分块大小的情况，Dask会自动合并分块直到读到换行符，极端情况下会出现单个分块包含全量文件内容的情况
• 监控任务运行状态：访问本地Dask仪表盘http://127.0.0.1:8787/status，观察Bytes stored per worker指标，确认是否存在单worker内存持续上涨到阈值后重启的情况

1. 调整读取参数优化分块 #

优先修改read_text的配置，强制指定分块大小，同时开启行分隔校验：

# 可根据实际单条JSON大小调整分块值，避免分块自动合并
b = db.read_text('2019-12-16-latest-level.json', blocksize='128MB', linedelimiter='\n').map(json.loads)

2. 调整集群内存配置 #

放宽worker内存阈值，避免频繁重启：

# 初始化集群时调整内存相关参数
cluster = LocalCluster(
    memory_limit='30GiB', # 单worker内存上限，预留部分内存给系统进程
    memory_target_fraction=0.8, # 内存用到80%时开始 spill 到磁盘
    memory_spill_fraction=0.9, # 用到90%强制spill
    memory_pause_fraction=0.95, # 用到95%暂停任务，而不是直接重启
)
client = Client(cluster)

3. 优化head调用逻辑 #

df.head()默认会尝试加载前1000行数据，如果前N个分块都没读到足够行数，会持续加载后续分块，极端情况会触发全量数据加载。可以指定npartitions参数只加载前几个分块：

# 只加载前2个分块的数据取head，避免扫描全表
df.head(n=10, npartitions=2)

4. 预处理大文件拆分 #

如果单JSON文件确实超过单worker内存上限，提前将大文件拆分为多个小的行分隔JSON文件，再用Dask加载：

# 用split命令拆分大文件，每个文件100万行
split -l 1000000 2019-12-16-latest-level.json split_level_

读取拆分后的所有文件即可解决单分块过大问题。

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