在KubeRay上计算Dask数组时，如何限制并行任务数量？

针对你遇到的Ray一次性提交大量任务导致OOM、现有CPU资源浪费的问题，以下是几种更优的解决方法：

1. 调整Dask全局配置，精准控制并行度 #

直接通过Dask配置限制全局并行任务数，并为每个Ray Worker分配刚好匹配任务需求的CPU资源，避免闲置浪费：

import dask

# 配置每个Ray Worker使用1核（匹配单个任务的CPU需求）
# 同时限制全局最大并行任务数为1000，避免任务过载
dask.config.set({
    "ray.worker.num_cpus": 1,
    "ray.scheduler.max_concurrent_tasks": 1000
})

# 后续创建数据立方体、计算均值的代码保持不变
cube = create_cube(size=...)
mean = cube.mean(axis=(0,1,2)).compute()

这种方式让1000个Worker各占1核，充分利用集群CPU资源，同时限制并行任务数，避免内存溢出。

2. 启用Dask任务批处理，控制任务提交节奏 #

通过设置任务批处理大小，让Ray分批次提交任务，而非一次性推送全部233280个任务，避免集群瞬间过载：

from dask_ray import ray_delayed

# 替换原有的@dask.delayed注解，启用批处理
@ray_delayed(batch_size=1000)
def _delayed_func(x: int, y: int, z: int) -> np.array:
    return np.random.randn(1000,1000,1000)

# 或者全局设置批处理大小
# dask.config.set({"ray.batch_size": 1000})

批处理会让Ray每次仅提交指定数量的任务，集群可以平稳启动Worker并逐步处理，配合自动扩缩容也不会出现内存突增的情况。

3. 通过KubeRay CRD限制集群最大Worker数 #

直接在OpenShift的KubeRay集群配置中，限制Worker的最大数量和单Worker资源配额，从集群层面控制并行度：

# RayCluster CRD示例配置
apiVersion: ray.io/v1alpha1
kind: RayCluster
metadata:
  name: dask-ray-cluster
spec:
  headGroupSpec:
    # Head节点配置...
  workerGroupSpecs:
  - replicas: 0
    maxReplicas: 1000  # 限制Worker最大数量为1000
    template:
      spec:
        containers:
        - name: ray-worker
          resources:
            limits:
              cpu: "1"  # 单Worker分配1核
              memory: "8Gi"  # 单Worker内存匹配单个Chunk需求
            requests:
              cpu: "1"
              memory: "8Gi"

这种方式从基础设施层面锁定资源上限，确保即使自动扩缩容也不会超出集群承载能力，同时每个Worker的资源刚好匹配任务需求，无资源浪费。

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