Dask Dataframe执行compute操作致集群内核无预警重启问题问询

• 核心问题不是集群总内存不足，是你调用df['Y'].compute()时，会把该列的全量数据从所有worker节点拉取到运行Python内核的驱动节点本地内存，而非在集群侧完成计算后返回结果。你的Y列全量数据体积远大于驱动节点（即你运行Notebook/脚本的进程）的可用内存，触发驱动进程OOM被系统直接杀死，表现就是内核无预警重启、所有已定义变量失效，集群侧也会因为驱动断连触发重启。
• 额外诱因是分区数设置不合理：共3132个分区对应83个worker、332个总线程，单worker平均承载近38个分区，分区碎片化严重，调度开销大，若存在数据倾斜还会导致单个worker内存被打满，加剧异常概率。另外persist()是异步执行操作，若你未等待数据完全加载到集群内存就执行后续计算，也可能触发调度拥堵。
• 常见认知误区：数据persist到集群内存不代表compute操作不会拉取全量数据到本地，只有聚合类操作（比如求和、算分位数）的compute才会只返回小体积的统计结果，直接对原始列/表调用compute会拉取全量原始数据。

第一步：优化分区与持久化逻辑 #

先把分区数调整到和总线程数匹配的合理区间（建议为总线程数的1_{2倍，即332}664个分区），再执行持久化，等待持久化完全完成后再做后续操作：

from dask.distributed import wait
# 重分区减少碎片化，降低数据倾斜概率
df = df.repartition(npartitions=332).persist()
# 阻塞等待所有数据持久化到集群内存完成
wait(df)

第二步：无需拉取全量数据即可完成箱线图绘制与winsorize处理 #

箱线图绘制、winsorize处理都只需要分位数统计值，完全不需要把全列数据拉到本地：

1. 分布式计算所需分位数，仅把统计结果返回本地：

   # 按需调整分位点，winsorize常用1%/99%或5%/95%截断
   qs = df['Y'].quantile([0, 0.01, 0.25, 0.5, 0.75, 0.99, 1]).compute()
   lower_cap = qs.loc[0.01]
   upper_cap = qs.loc[0.99]
   

2. 分布式执行winsorize截断，全程数据不离开集群节点：

   # 按分位数截断异常值，元数据根据Y列实际类型调整即可
   df['Y_winsorized'] = df['Y'].map_partitions(
       lambda col: col.clip(lower=lower_cap, upper=upper_cap),
       meta=('Y', 'float64')
   )
   

3. 箱线图绘制不需要全量数据，抽取1%以内的代表性样本拉到本地即可，样本量完全可以支撑分布展示，不会撑爆本地内存：

   # 固定随机种子保证结果可复现
   y_sample = df['Y'].sample(frac=0.005, random_state=42).compute()
   y_sample.plot.box()
   

注意：除非你确认目标数据体积远小于驱动节点的可用内存，否则永远不要直接对Dask的原始DataFrame/列直接调用无聚合的compute()，这类操作会把全量分布式数据拉到本地，极容易触发OOM。

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