Ray中Object Spilling问题：大量图片处理时性能骤降的解决咨询

你遇到的是Ray对象存储溢出触发磁盘交换（Spill）的问题。当一次性提交10000个任务并调用ray.get()获取所有结果时，所有图片的numpy数组会被一次性加载到Ray的对象存储中，当数据总量超过对象存储内存上限（或系统实际可用内存）时，Ray会将部分数据写入磁盘——磁盘IO速度远慢于内存，直接导致耗时剧增。

以下是几种可行的优化方案：

1. 分批处理任务，避免一次性加载所有结果 #

不要一次性提交所有任务并获取结果，分批次处理，每批处理完成后释放内存再处理下一批：

import ray
import numpy as np
from PIL import Image

ray.init(
    object_store_memory=20 * 1024 * 1024 * 1024,  # 根据系统可用内存调整，例如设为20GB
    ignore_reinit_error=True,
    num_cpus=128,
    num_gpus=1,
)

img_paths = np.array([...])  # 你的200k图片路径
batch_size = 1000  # 每批处理的数量

@ray.remote
def read_img(path):
    img = np.asarray(Image.open(path))
    return img

all_images = []
for i in range(0, 10000, batch_size):
    batch_paths = img_paths[i:i+batch_size]
    futures = [read_img.remote(path) for path in batch_paths]
    batch_images = ray.get(futures)
    all_images.extend(batch_images)
    # 显式清理当前批次对象，加速内存回收
    del futures, batch_images

2. 使用ray.wait控制并发结果数量 #

通过ray.wait控制同时加载到内存的任务结果数量，避免内存瞬间占满：

futures = [read_img.remote(path) for path in img_paths[:10000]]
all_images = []
remaining = futures.copy()

while remaining:
    # 每次获取100个完成的任务结果
    done, remaining = ray.wait(remaining, num_returns=100)
    batch_images = ray.get(done)
    all_images.extend(batch_images)
    del done, batch_images

3. 优化对象存储内存设置 #

你当前设置的object_store_memory=1000 * 1024 * 1024 * 100（约100GB）可能远超系统实际可用内存，Ray会自动将其限制为系统内存的约30%（默认行为）。需根据系统实际可用内存调整，比如系统有64GB可用内存时，可设为40 * 1024 * 1024 * 1024（40GB），确保有足够内存容纳批量数据，同时不占用系统全部内存。

4. 轻量化处理图片数据 #

如果不需要完整高清图片数组，可在任务中对图片压缩、缩放或提取必要特征，减少返回对象大小：

@ray.remote
def read_img(path):
    with Image.open(path) as img:
        # 缩放图片到指定尺寸，例如256x256
        img_resized = img.resize((256, 256))
        img_array = np.asarray(img_resized)
        # 可选：转换为uint8类型进一步压缩体积
        return img_array

5. 使用Ray Dataset处理大规模数据 #

Ray Dataset专为大规模批量数据设计，会自动处理内存管理、分批和并行，比手动提交任务更高效：

import ray
from ray.data import read_images

ray.init(
    ignore_reinit_error=True,
    num_cpus=128,
    num_gpus=1,
)

# 直接读取图片路径列表，Dataset自动并行处理
ds = read_images(img_paths[:10000])
# 自动分批加载并转换为numpy数组列表
all_images = ds.take_all()

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