Introductions To Ray

Ray란 무엇입니까

Ray는 AI 처리 및 Python Application의 확장을 위한 통합 프레임워크입니다. 이는 오픈 소스로 되어 있어서 자유롭게 사용이 가능합니다. Ray는 핵심 분산 런타임 기능과 머신러닝 연산을 간소화하는 AI 라이브러리 세트로 구성되어 있습니다. 제가 이 기능을 눈여겨 본 이유는 많은 기업에서 엔터프라이즈 환경에서 구동하고 있는 것도 있습니다. 그리고 Learning, Tuning, Serving 등의 기능을 Ray 안에서 할 수 있다는 점이 매력적으로 다가 왔습니다. 그리고 분산과 확장에 어느 정도 신경쓴 기능도 존재합니다. 한마디로 정리하면 Machine Learning 기능에 특화된 작은 Spark 같다는 개인적인 느낌이 들었습니다. 그래서 저는 이 페이지를 통해서 정말 간단한 Ray 설치와 실행 그리고 간단하게 Task를 발행해서 사용할 수 있는 준비를 해보고자 합니다.

Ray 설치와 실행

Ray는 Python 기반 위에 실행하기 때문에 설치 및 실행 하려면 기본적으로 python 설치 및 환경 세팅이 되어 있어야 합니다. 저는 Ubuntu기반에서 사용과 테스트를 했기 때문에 기본적으로 Python이 되어 있습니다.

Python pip를 통한 설치는 "pip install -U ray[options...]" 해당 방식으로 설치 할 수 있습니다. 설치 이후에 실행하는 것도 가능한데요. 저는 설치 결과를 아래 이미지 처럼 확인 했습니다

설치하면 뭔가 많이 설치하는 과정이 지나가면서 잘 될겁니다. 그러면 아래 그림과 같이 ray start --head 명령어로 실행 할 수 있습니다. 원래는 address 설정을 해야 하지만 우리는 local에서 테스트를 위해서 실행을 시키기 때문에 그냥 실행 시키겠습니다. --head를 사용하면 local 구성에 맞게 아래와 같이 실행이 됩니다.

Ray는 기본적으로 현재 상태나 로그 등을 Web UI를 통해 확인할 수 있도록 제공합니다. 따라서 실행을 하고 결과를 확인하는 간단한 방법은 Web UI를 통해서 확인 하는 것입니다. 8265포트에 우리가 설정한 대로 local로 접근할 수 있습니다. 실제로 확인을 먼저 하겠습니다.

여기 까지 확인하면은 Ray 설치 및 실행은 완료한 것이고 실제로 사용할 준비가 다 된 것입니다. 그 다음에는 코드를 실행시켜서 Ray에게 Task를 직접 실행시켜 보도록 하겠습니다.

Ray는 다양한 언어를 통해 사용할 수 있지만 가장 기반이 되는 Python을 통해서 명령을 수행하도록 하겠습니다. Task는 Ray에서 함수 호출을 Cluster에서 실행하는 것입니다. 쉽게 생각하면 원격 실행 함수라고 할 수 있습니다. 코드는 간단합니다. @ray.remote 어노테이션을 통해서 원격으로 실행할 Task를 정의 합니다. 그리고 ray.init()과 ray.get()을 통해서 Ray를 사용할 준비를 하고 실행 해서 결과를 얻게 할 수 있습니다. 테스트 환경에서 ray.init할 때에 인자로 ignore_reinit_error를 전달하면 여러 번 실행해도 조금 편하게 중복 init에 걸리지 않고 테스트 할 수 있습니다. 코드로 표현하면 아래와 같습니다.

        
            import ray.data

            @ray.remote
            def test(value):
                return value * value

            if __name__ == "__main__":
                ray.init(ignore_reinit_error=True)

                tasks = [test.remote(ii) for ii in range(10)]
                print(ray.get(tasks))

                ray.shutdown()
        
    

Ray를 통해 결과를 얻었습니다. 그리고 이 결과를 Web UI에서도 확인을 한다면 크로스 체크 상 문제가 없을 것으로 보입니다.

Ray의 WebUI에서는 Task 실행을 확인 로그는 Job Tab에서 확인 할 수 있습니다. 따라서 해당 탭에 정상적으로 표시가 되고 있는 걸로 보아 처음에 실행이 잘 된것으로 보이네요.

다음은 Actor 방식을 통해서 동일한 기능을 실행 해 보겠습니다. Actor가 가지는 가장 큰 특징은 Task와는 달리 State를 가진 다는 점입니다. 그래서 마치 Worker처럼 단발성이 아니라 StateFul한 방식으로 데이터를 쌓거나 캐시 하면서 처리할 수 있게 합니다. Python에서 가장 간편하게 하는 방식은 Class에 데이터를 누적시키는 방식이라고 생각합니다. 따라서 해당 방식으로 구현해 보겠습니다.

        
            import ray.data

            @ray.remote
            class Counter:

                def __init__(self):
                    self.value = 0

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

                    return self.value

            if __name__ == "__main__":
                
                ray.init(ignore_reinit_error=True)

                counter = Counter.remote()
                print(ray.get([counter.increment.remote() for _ in range(20)]))

                ray.shutdown()
        
    

이것도 동일하게 WebUI에서 확인하겠습니다.

마지막으로 저의 백엔드 서버에서 Ray와 통신할 수 있는 기능을 테스트 해보겠습니다. 이 기능은 Ray의 serve라는 API 기능을 통해서 활용할 수 있습니다. 그리고 Python 기반이기 때문에 FastAPI를 활용하도록 하겠습니다. FastAPI는 Python 기반의 Web Backend Framework 입니다. Python에서는 Django, Flask 등의 Web Framework도 존재 하는데요. FastAPI를 사용하는 이유는 1) Async 처리를 사용하기 쉽고 2) Django나 다른 Framework에 비해 방대하지 않다는 점 때문 입니다. 저는 Django로도 사이드 프로젝트도 해보고 배포도 했습니다만은 이 점 때문에 FastAPI가 좀 더 사용하기 낫다는 개인적인 의견입니다. @serve.deployment를 통해서 배포할 수 있도록 하고 @serve.ingress를 통해서 FastAPI의 http 프로토콜과 연결하도록 하겠습니다.

        
            from typing import List
            from fastapi import FastAPI
            from pydantic import BaseModel
            import ray
            from ray import serve

            # Fast API 사용할 수 있도록 하는 지점입니다.
            app = FastAPI()

            # Input 형태(여기서는 http request의 인자 중 하나)
            class InferRequest(BaseModel):
                x: List[float]

            @serve.deployment(ray_actor_options = {"num_cpus": 1})
            @serve.ingress(app)
            class InferenceService:

                def __init__(self):
                    # 여기서 모델 로딩/워밍업 등을 수행
                    # self.model = load_model(...)
                    pass

                @app.get("/health")
                async def health(self):
                    # service의 상태 정보인데 일단 무조건 ok
                    return {"status": "ok"}

                @app.post("/infer")
                async def infer(self, req: InferRequest):
                    # 실제 추론 혹은 처리 로직

                    y = sum(req.x)
                    return { "y": y }

            if __name__ == "__main__":
                
                ray.init()
                serve.run(InferenceService.bind(), route_prefix="/")

                # serve HTTP 기본 포트는 8000
                print("Serving on http://127.0.0.1:8000")
                print("Try: GET /health, POST /infer")

                # 개발 중에는 블로킹을 걸어 둬야 코드에서 console로 결과를 확인할 수 있습니다.
                import time
                while True:
                    time.sleep(18000)
        
    

이런 식으로 하면 아직 서비스 하기에는 많이 부족하긴 합니다만 개인 Backend에서도 Ray를 사용해서 Task 및 Actor를 실행 할 수 있습니다. 사실 실제 서빙을 염두해 두고 무언가를 처리하길 원한다면 마지막 연동 기능이 기초이자 핵심일 것으로 보입니다. 서버 끼리 원활하게 소통하고 처리해야 진짜 라이브 서비스에서 적용할 수 있는 발판이 될 테니까요

다음에는 Product Level에서 필요한 다양한 기능을 소개 하고 간단한 기능을 서빙하는 데 까지 포스팅을 할 수 있도록 노력하겠습니다. 그것이 제가 Ray를 이해하는 목표 이니까요