레이 (분산 컴퓨팅)

레이(Ray)는 머신 러닝 및 분산 애플리케이션을 위한 범용 분산 컴퓨팅 프레임워크이다.

개요

레이는 파이썬 중심의 API를 기반으로 하며, 간결한 코드로 대규모 분산 처리를 가능하게 해주는 범용 분산 실행 엔진이다. Ray는 동시성(concurrency), 병렬성(parallelism), 분산성(distribution)을 모두 지원하며, 특히 강화 학습, 하이퍼파라미터 튜닝, 대규모 데이터 처리 등의 머신 러닝 워크로드에 적합하다. 서버리스(serverless) 방식의 분산 실행 모델을 채택하여 사용자가 인프라를 직접 관리하지 않고도 작업을 분산 실행할 수 있도록 설계되었다.

주요 특징

• 파이썬 기반의 간단한 API 제공
• 함수 기반의 원격 실행(Remote Function) 및 액터(Actor) 모델 지원
• 스케일 아웃 가능한 병렬 처리 프레임워크
• 다양한 머신 러닝 툴킷과 통합(TensorFlow, PyTorch 등)
• 자동 리소스 관리 및 스케줄링 기능 포함
• 강화 학습, 서빙, 튜닝을 위한 하위 프레임워크 포함(Ray RLlib, Ray Serve, Ray Tune)

주요 구성 요소

• Ray Core: 분산 태스크 및 액터 실행을 위한 기본 모듈
• Ray Tune: 대규모 하이퍼파라미터 튜닝을 위한 라이브러리
• Ray RLlib: 강화 학습 알고리즘의 분산 학습을 위한 프레임워크
• Ray Serve: 머신 러닝 모델 서빙을 위한 확장 가능한 서버리스 솔루션
• Ray Data: 분산 데이터 로딩 및 전처리를 위한 컴포넌트

아키텍처

레이 아키텍처

Ray 클러스터는 헤드 노드(head node)와 하나 이상의 워커 노드(worker node)로 구성된다.

헤드 노드

헤드 노드는 클러스터 전체를 관리하며, 워커 노드와 동일한 구성 요소에 더해 아래의 두 가지 주요 컴포넌트를 추가로 포함한다.

• Global control store (GCS): 클러스터 전역 정보를 저장하는 구성 요소로, 객체 테이블, 태스크 테이블, 함수 테이블, 이벤트 로그 등을 포함한다. 이 정보는 웹 UI, 디버깅, 프로파일링 등에 활용된다.
• Autoscaler: 워크로드에 따라 워커 노드를 자동으로 생성하거나 제거하여 자원 낭비를 줄이고 성능을 최적화한다.

헤드 노드는 기본적으로 단일 장애 지점(single point of failure)으로, 장애 발생 시 클러스터 전체가 소실되며 재생성이 필요하다. 이로 인해 기존 워커 노드는 고아 노드(orphaned node)가 되어 수동으로 제거해야 할 수 있다.

워커 노드

각 워커 노드는 Raylet이라는 핵심 컴포넌트를 포함하며, 이는 다음 두 구성 요소로 구성된다.

• Object store: 분산 캐시와 유사하게 작동하며, 클러스터 내 모든 object store는 연결되어 하나의 통합 캐시를 구성한다.
• Scheduler: 각 노드는 로컬 스케줄러를 통해 작업을 수행하며, 다른 노드와 통신하여 전역 분산 스케줄러로 기능한다.

리모트 함수

Remote function은 Ray에서 분산 병렬 처리를 위해 사용하는 기본 단위로, 다음과 같은 현대 애플리케이션 요구사항을 충족한다.

• 동일한 코드를 여러 코어나 머신에서 실행 가능
• 오류 발생 시 자동 재시도 등 내결함성 제공
• 큰 파라미터도 효율적으로 처리 가능
• 프로세스 간 정보 전달이 용이함

Ray는 함수 호출을 올바른 프로세스로 자동 매핑하여 실행하며, 함수 호출 결과는 곧바로 ObjectRef 형태로 반환된다. ObjectRef란 실제 결과가 저장될 미래 객체에 대한 일종의 포인터이다.

실행 방식

• 사용자가 remote 함수를 호출하면, Ray는 작업을 비동기적으로 분산 실행한다.
• ObjectRef를 통해 값을 나중에 받아올 수 있으며, ray.get()을 사용하여 결과를 확인한다.
• remote 함수는 반드시 다른 머신에서 실행되는 것은 아니며, 같은 노드 내 다른 프로세스일 수도 있다.

예시 코드

import ray
ray.init()

@ray.remote
def f(x):
    return x * x

futures = [f.remote(i) for i in range(4)]
print(ray.get(futures))  # [0, 1, 4, 9]

동작 흐름

• 함수 호출자는 해당 작업의 소유자(owner)가 되며, ObjectRef를 관리한다.
• 작업 제출 시, owner는 인자로 전달된 모든 ObjectRef가 준비될 때까지 대기한다.
• 준비 완료되면, 분산 스케줄러에 자원을 요청하고 작업 실행을 위한 워커 주소를 받는다.
• gRPC를 통해 워커에게 작업 명세를 전달하며, 워커는 작업을 실행 후 결과를 저장한다.
  • 작은 결과는 인라인으로 전달되고, 큰 객체는 local object store에 저장된다.

오류 처리

• 응용 수준 오류: 예외가 발생하면 해당 예외 객체가 결과로 반환되며, 재시도는 이루어지지 않는다.
• 시스템 수준 오류: 예를 들어 워커 프로세스가 죽는 경우, 지정된 횟수만큼 자동 재시도된다.

리모트 액터

Remote function은 상태를 갖지 않는(stateless) 함수의 병렬 실행에 적합하다. 그러나 상태를 유지해야 하는 경우에는 Remote Actor를 사용한다.

개요

• 상태를 유지하는 객체를 분산 환경에서 사용할 수 있게 해주는 기능이다.
• @ray.remote 데코레이터를 클래스에 붙여 사용한다.
• 액터는 하나의 프로세스로 실행되며, 그 안의 상태는 외부에서 접근할 수 없다.
• 상태 변경은 해당 액터에 메시지를 보내는 방식으로만 가능하다.

예시 코드

import ray
ray.init()

@ray.remote
class Counter(object):
    def __init__(self):
        self.n = 0

    def increment(self):
        self.n += 1

    def read(self):
        return self.n

counters = [Counter.remote() for i in range(4)]
[c.increment.remote() for c in counters]
futures = [c.read.remote() for c in counters]
print(ray.get(futures))  # [1, 1, 1, 1]

액터 모델

• 액터는 고유 주소(handle)를 가진 프로세스이다.

• 내부 상태는 외부에서 직접 접근할 수 없으며, 메시지를 통해 간접적으로만 조작할 수 있다.
• 각 액터는 메시지를 하나씩 순차적으로 처리하므로 상태 충돌이 발생하지 않는다.
• 상태를 나누거나 복제할 수 있다면, 여러 액터 풀을 구성하여 처리량을 향상시킬 수 있다.

동작 방식

• Ray는 액터를 새로운 워커로 생성하고, 그 이후 모든 메서드는 해당 워커에서 실행된다.
• 액터 생성은 GCS에 등록되어 메타데이터(IP, 포트 등)를 공유한다.
• 각 클라이언트는 GCS를 재조회하지 않고, 캐시된 메타데이터로 직접 gRPC를 통해 액터에게 메시지를 전송한다.
• 메서드 호출은 remote function과 동일하게 ObjectRef를 반환한다.

장애 처리

• 액터는 상태를 가지므로 복구가 더 복잡하다.
• 메시지 처리 중 실패 시 자동 재시도되지 않는다.
• 메시지 사이에서 실패한 경우, Ray는 다음 호출 시 최대 설정된 횟수만큼 재시도한다.
• 초기화 실패도 첫 메시지 실패와 동일하게 처리된다.

장애 허용

Ray는 애플리케이션 계층과 시스템 계층으로 구성되어 있으며, 두 계층 모두 장애 복구 기능을 갖추고 있다.

시스템 계층 구성 요소

• Global Control Store (GCS): Ray의 전체 상태를 저장하는 중앙 구성 요소
• 분산 스케줄러: 작업 분배 및 자원 할당을 담당
• 분산 오브젝트 저장소: 객체 데이터를 클러스터 내에서 분산 관리

Global Control Store (GCS)

• GCS는 시스템 상태를 중앙에서 유지하여 나머지 구성 요소들이 무상태(stateless)로 설계되도록 한다.
• 구성 요소가 장애로 인해 재시작되면, GCS에서 상태를 읽어와 복구한다.
• 이를 통해 객체 저장소와 스케줄러를 독립적으로 확장할 수 있다.
• 기본적으로 GCS는 헤드 노드에 존재하며, 단일 장애 지점(SPOF)이 된다.

Remote Function의 장애 허용

• Remote function은 상태가 없기 때문에 복구가 간단하다.
• 시스템 오류로 실패한 작업은 지정된 최대 횟수만큼 자동 재시도된다.
• 응용 오류(Exception 발생)는 자동 재시도되지 않고, 예외 객체로 반환된다.
• 최대 재시도 횟수는 `@ray.remote(max_retries=n)`으로 설정 가능하다.

Remote Actor의 장애 허용

• 액터는 상태를 갖기 때문에 복구가 더 복잡하다.
• 초기화, 메시지 처리 중, 요청 간 등 어떤 단계에서도 실패할 수 있다.

메시지 처리 중 실패

• 자동 재시도되지 않음
• 다음 메시지를 처리할 때까지 대기하며, 최대 max_restarts 횟수까지 액터를 재시작한다.

요청 간 실패

• 다음 메시지 호출 시 자동으로 액터를 복구한다.
• 상태 복구 로직이 적절히 구현되어 있다면 실패는 느린 처리 외에는 큰 영향을 주지 않는다.
• 상태 복구 로직이 없을 경우, 액터는 초기 상태로 재시작된다.

기타 사항

• 대부분의 자원은 애플리케이션 종료 시 자동 정리된다.
• Detached 리소스(예: detached actor, placement group)는 클러스터가 유지되는 한 계속 유지된다.

기타 설계 사항

직렬화(Serialization)

• Ray는 Plasma in-memory object store를 사용하여 객체를 노드 간 및 프로세스 간 효율적으로 전달한다.
• NumPy 배열은 zero-copy 방식으로 같은 노드 내 여러 워커가 공유할 수 있다.
• Plasma에 저장된 객체는 불변이며(shared memory)에 저장된다.
• 객체는 요청이 있을 때에만 다른 노드로 전송되며, 사전 복제되지 않는다.

리소스 관리

• 기본적으로 함수와 액터는 동일한 자원(CPU 1개)을 요구한다.
• GPU, 메모리 등 필요한 자원을 명시할 수 있으며, 스케줄러는 이를 고려하여 실행 노드를 할당한다.
• 자원이 부족한 경우, 오토스케일러가 필요한 사양을 가진 노드를 추가한다.

자동 스케일링(Autoscaler)

• 오토스케일러는 다음을 담당한다:
  • 워커 노드 생성
  • 워커 노드 제거
  • 워커 재시작

배치 전략(Placement Groups)

• Placement group은 작업과 자원을 함께 배치하기 위한 논리적 단위이다.
• 사전 자원 할당(preallocation)을 통해 작업 지연을 줄일 수 있다.
• Pack: 하나의 노드에 최대한 묶어 배치하여 locality 향상
• Spread: 여러 노드에 분산 배치하여 신뢰성 및 부하 분산 확보

네임스페이스(Namespaces)

• Ray는 작업과 액터를 논리적으로 그룹화하기 위해 네임스페이스를 사용한다.
• 기본적으로 익명 네임스페이스에서 실행된다.
• 여러 프로그램에서 동일한 액터를 공유하려면 동일한 네임스페이스를 명시해야 한다.

ray.init(namespace="bdad")

의존성 관리

• Conda 또는 virtualenv 기반으로 동적 환경 구성이 가능하다.
• 런타임 환경을 함수 단위로 설정할 수도 있다.

ray.init(runtime_env={"pip": "requirements.txt"})

@ray.remote(runtime_env={"conda": ["some_package"]})
def some_function(x): ...

Job API

• 클러스터에 작업을 제출하고 추적할 수 있는 API를 제공한다.
• 작업 ID로 상태를 조회하거나 실행 로그를 확인할 수 있다.

예시 코드

import ray

ray.init()

@ray.remote
def f(x):
    return x * x

futures = [f.remote(i) for i in range(4)]
print(ray.get(futures))

위 예시는 4개의 태스크를 병렬로 실행한 뒤 결과를 수집하는 단순한 병렬 처리 예제이다.

사용 사례

• 대규모 강화 학습 환경 구축 및 실험(RLlib)
• 수천 개의 하이퍼파라미터 조합을 통한 모델 최적화(Ray Tune)
• 실시간 모델 예측 요청 처리(Ray Serve)
• 분산 데이터 로딩 및 전처리(Ray Data)
• 일반적인 병렬 처리 및 배치 작업 수행

Ray와 다른 프레임워크 비교

• Dask와 유사한 파이썬 중심의 API를 제공하지만, Ray는 액터 모델 기반의 상태 유지 작업에 더 적합하다.
• Apache Spark보다 경량이며, 머신 러닝에 특화된 구성 요소를 포함하고 있다.
• Celery와 달리 분산 스케줄링 및 리소스 관리를 자동으로 처리한다.

같이 보기

• 분산 처리
• 강화 학습
• 하이퍼파라미터 최적화
• 서버리스 컴퓨팅
• 파이썬

참고 문헌

• Moritz, Philipp et al. Ray: A Distributed Framework for Emerging AI Applications. Proceedings of the 13th USENIX Symposium on Operating Systems Design and Implementation (OSDI), 2018.
• Anyscale, Inc. The Ray Documentation. https://docs.ray.io/