Dask计算任务无终止仅单worker运行，如何分发到所有核心执行？

代码存在的核心问题 #

• 基础语法错误
  • 未执行client = Client(cluster)将客户端与PBS集群绑定，代码中直接使用的client变量未定义，任务提交逻辑本身就存在异常
  • PBSCluster的参数名书写错误：local directory应为下划线分隔的local_directory
• 任务逻辑无法并行，且存在OOM风险
  • 通过client.submit提交的是单个生成全量numpy数组的任务，该任务本身是不可拆分的单进程任务，只能在单个worker上执行
  • 你要生成的数组大小约为7.5GB（6230007341*4字节），单worker生成该数组时加上运行时内存开销很容易触发内存超限，导致worker被集群杀死、任务反复重启
  • 用da.from_delayed包装单块数组得到的Dask Array只有1个分区，后续所有计算都只能绑定到该分区所在的worker，完全无法利用多核心多节点资源
• 集群未配置伸缩规则
  • 初始化PBSCluster后没有调用cluster.scale()或cluster.adapt()指定worker数量，默认只会启动1个worker，就算任务可拆分也没有多余资源可用

修复方案 #

你要生成随机数组无需单独提交单任务，直接使用Dask Array原生的随机生成接口即可自动实现分块并行，修改后的代码如下：

import numpy as np
from dask.distributed import Client
from dask.jobqueue import PBSCluster
import dask.array as da

# 修正参数错误初始化集群
cluster = PBSCluster(
    processes=1, 
    cores=8, 
    memory="40GB", 
    local_directory='$TMPDIR',
    walltime='1:00:00', 
    interface='ib0'
)

# 按需调整worker数量，比如启动8个worker，共使用64核心
cluster.scale(8)
# 绑定客户端到集群
client = Client(cluster)

# 直接用Dask的随机接口生成数组，按第一维分块，每个块10000条，自动并行
arr = da.random.uniform(
    size=(623000, 73, 41), 
    chunks=(10000, 73, 41), 
    dtype=np.float32
)
# 计算任务会自动分发到所有worker的核心执行
res = arr.compute()

如果你的实际场景确实需要从自定义的生成逻辑转换为Dask Array，需要将生成逻辑拆分为多个小的delayed任务，每个任务对应数组的一个分块，再批量传入da.from_delayed，即可实现并行调度。

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