Ray
Ray는 AI를 위한 컴퓨팅 레이어입니다. ChatGPT 학습과 대규모 워크로드를 지원합니다. v3.0(2025)은 효율성을 개선하고 에이전트를 위한 MCP 서버를 추가했습니다.
사용 시기
- 분산 Python 애플리케이션: 단일 랩탑에서 대규모 클라우드 클러스터까지 컴퓨팅 집약적 Python 작업을 원활하게 확장합니다.
- Ray Train을 사용한 분산 ML 학습: 다중 노드 PyTorch, XGBoost, LightGBM 학습 작업을 오케스트레이션합니다.
- Ray Serve를 사용한 프로덕션 모델 서빙: 동적 자동 확장 기능으로 LLM 및 다중 모델 마이크로서비스 파이프라인을 서빙합니다.
- Ray Tune을 사용한 하이퍼파라미터 최적화: 수백 개의 워커에서 베이지안, Optuna, PBT(인구 기반 학습) 스윕을 실행합니다.
빠른 시작
import ray
ray.init()
# 원격 분산 태스크
@ray.remote
def square(x):
return x * x
# 클러스터 전체에서 태스크를 병렬로 실행
futures = [square.remote(i) for i in range(10)]
results = ray.get(futures)
print("Computed squares:", results)
핵심 개념
#분산 태스크 및 상태 저장 액터
클러스터 워커 전체에서 함수와 클래스를 확장합니다:
import ray
import time
# 로컬 초기화 또는 원격 Ray 클러스터에 연결
ray.init(ignore_reinit_error=True)
# 1. 무상태 원격 함수 (분산 태스크)
@ray.remote
def process_data_chunk(chunk_id: int) -> dict:
time.sleep(0.5) # CPU/IO 집약적 연산 시뮬레이션
return {"chunk_id": chunk_id, "processed_items": 1000 * chunk_id}
# 클러스터 전체에서 태스크를 동시에 실행
futures = [process_data_chunk.remote(i) for i in range(8)]
results = ray.get(futures) # 결과 수집
print("Processed Chunks:", results)
# 2. 상태 저장 원격 클래스 (분산 액터)
@ray.remote
class GlobalCounter:
def __init__(self):
self.count = 0
def increment(self) -> int:
self.count += 1
return self.count
counter_actor = GlobalCounter.remote()
inc_futures = [counter_actor.increment.remote() for _ in range(5)]
print("Final actor count:", ray.get(inc_futures[-1]))
#Ray Serve를 사용한 프로덕션 모델 서빙
동적 요청 배치 기능으로 확장 가능한 REST 엔드포인트 배포:
from ray import serve
import torch
from transformers import pipeline
@serve.deployment(num_replicas=2, ray_actor_options={"num_cpus": 1, "num_gpus": 0})
class SentimentClassifier:
def __init__(self):
self.classifier = pipeline("sentiment-analysis")
async def __call__(self, request):
json_data = await request.json()
text = json_data.get("text", "")
return self.classifier(text)
app = SentimentClassifier.bind()
# serve.run(app)
#Ray Tune을 사용한 분산 하이퍼파라미터 튜닝
병렬 하이퍼파라미터 실험 실행:
from ray import tune
def training_function(config):
# 시뮬레이션된 모델 학습
for step in range(10):
intermediate_score = config["alpha"] * step + config["beta"]
tune.report(loss=1.0 / (intermediate_score + 0.1))
search_space = {
"alpha": tune.uniform(0.1, 1.0),
"beta": tune.choice([1, 2, 5])
}
tuner = tune.Tuner(
training_function,
param_space=search_space,
tune_config=tune.TuneConfig(num_samples=10, metric="loss", mode="min")
)
results = tuner.fit()
print("Best hyperparameters:", results.get_best_result().config)
일반적인 패턴
모델 서빙을 위한 상태 저장 분산 액터
문제: 함수 호출 시마다 대용량 모델 가중치를 메모리에 다시 로드해야 합니다.
해결책: Ray 액터를 사용하여 요청 간에 인메모리 상태를 유지합니다:
@ray.remote(num_gpus=1)
class ModelServer:
def __init__(self):
# GPU VRAM에 가중치를 한 번 로드
self.model = load_large_model()
def predict(self, sample):
return self.model(sample)
server_actor = ModelServer.remote()
prediction = ray.get(server_actor.predict.remote(sample_data))
모범 사례 (2026)
- 권장: Ray Dashboard(포트 8265)를 확인하여 노드 CPU/GPU 사용률, 액터 배치, 객체 저장소 메모리를 모니터링하세요.
- 권장: 반복적인 직렬화를 방지하기 위해 Ray 플라즈마 객체 저장소(
ray.put(data))를 통해 대용량 읽기 전용 데이터셋을 전달하세요. - 권장: 데코레이터(
@ray.remote(num_cpus=2, num_gpus=1))에서 리소스 요구 사항을 명시적으로 지정하세요. - 권장: Ray Serve를 사용하여 내장된 자동 배치 기능으로 프로덕션 다중 모델 마이크로서비스를 구축하세요.
- 비권장: 원격 태스크 내부에서
ray.get()을 호출하지 마세요. 파이프라이닝을 유지하기 위해 ObjectRef를 다른 원격 태스크에 직접 전달하세요. - 비권장: 수백만 개의 미세 태스크를 생성하지 마세요. 스케줄러 오버헤드를 최소화하기 위해 세분화된 태스크를 함께 배치하세요.
- 비권장: 데이터베이스 연결이나 열린 파일과 같은 대용량 상태 저장 객체를 원격 인수로 전달하지 마세요.
문제 해결
| 오류 | 원인 | 해결 방법 |
|---|---|---|
RaySystemError: Object store full | 할당된 공유 메모리 플라즈마 저장소보다 더 많은 데이터를 생성하는 태스크입니다. | ray.get()을 사용하여 결과를 소비하거나, 대용량 배열을 디스크 저장소에 저장하세요. |
Actor died unexpectedly | 처리되지 않은 예외 또는 OOM으로 인해 워커 프로세스가 충돌했습니다. | Ray Dashboard(http://localhost:8265)에서 로그를 확인하고 워커 RAM을 늘리세요. |
Task failed to schedule: Unsatisfiable resource request | 클러스터에서 사용 가능한 것보다 더 많은 CPU/GPU(num_gpus=2)를 요청했습니다. | ray.cluster_resources()를 통해 사용 가능한 리소스를 확인하세요. |