ray를 사용해보자

파이썬으로 병렬 처리를 할 수 있는 대표적인 방법으로는 multiprocessing 라이브러리를 사용하는 방법이 있는데요. 

사용해보신 분들은 아시겠지만, multiprocessing라이브러리는 개념이 어렵고, 코드도 꽤나 복잡합니다. 

그래서 초기 진입장벽이 높은 편인데요. 그렇다면 쉽게 병렬 처리를 할 수 있는 방법은 없을까요?

 

바로 ray라는 라이브러리를 사용하시면 아주 쉽게 병렬처리를 할 수 있습니다. 

 

Ray란?

ray는 파이썬 병렬처리 라이브러리중 하나이며, 기존 multiprocessing 라이브러리와 비교했을 때 많은 장점을 가지고 있습니다. 

multiprocessing은 적용을 위해서 기존의 코드를 많이 바꾸어야하고, 코드 이해가 쉽지 않습니다. 반면 ray는 정말 단순하기 때문에 코드를 따로 고칠 필요가 없고, 굉장히 빠르다는 장점이 있습니다. 

 

실제로 병렬처리를 하지 않은 코드, multiprocessing, ray를 이용했을 때 걸린 시간은 아래와 같습니다. 

ray가 압도적으로 빠른 것을 알 수 있습니다. 

 

마지막으로 ray, multiprocessing의 특징을 표로 정리하고 사용법을 알아보도록 합시다. 

multiprocessing	ray
- '프로세스 스포닝'이라는 것을 지원 ㄴ 부모 프로세스가 자식 프로세스를 생성하는 것 - 프로세스 풀을 제어하는 Pool 객체를 통해 병렬처리 - Pool 클래스 수정해서 병렬처리 적용	- 별도의 코드 수정이 필요 없음 - 머신러닝/딥러닝을 위해 개발되었음 ㄴ하지만 다양하게 활용이 가능함 - 다양한 환경에서 사용이 가능함  ㄴ로컬, 쿠버네티스 등

 

Ray의 구성 요소

ray를 사용하기 전 ray가 어떻게 구성되어있는지 알아야합니다. 

ray는 크게 Task, Object, Actor로 이루어져있습니다. 

 

1. Task

• Task는 호출자와 다른 프로세스에서 실행되는 Single Function Call입니다.
• 호출자와 비동기적으로 실행되며, remote function을 이용하여 표기합니다. 
• 쉽게 말하자면 "어떤 행동을 병렬 처리할건데?"를 정의해준다고 생각하시면 됩니다. 
  • 병렬처리를 수행할 함수 위에 @ray.remote를 붙여주면 됩니다.
  • 호출 시에는 함수이름.remote()라고 하시면 됩니다. ex) mul.remote()

@ray.remote
def mul(x):
	return x*10

2. Object

• Task를 통해 실제로 반환되는 결과값이나, ray.put()을 통해서 생성되는 값입니다. 
• Ray 시스템의 Object Store이라는 곳에 저장됩니다. 
• 불변하는 데이터입니다. 
• ray.put()이 반환하는 ObjectRef를 통해서 참조할 수 있습니다. 
• 쉽게 말하자면 입력, 출력되는 값을 담아두는 공간이라고 생각하시면 됩니다. 
  • 입력할 값을 담기 위해서는 ray.put()함수를 이용합니다. ex) ray.put(arr)

arr = np.random.random(10000)
arr = ray.put(arr)

3. Actor

• 워커 프로세스입니다. 
• 위에서 생성한 Object들을 실제 Task에 담아 실행시키는 역할을 합니다. 

result = ray.get(mul.remote(arr))

 

Ray 사용법

ray를 사용하기 앞서 설치부터 해줘야겠죠?

pip install ray

 

병렬처리를 할 함수를 정의해줍니다. 

이 때, @ray.remote를 반드시 붙여줘야합니다. 

@ray.remote
def square(x):
    return x**2

 

ray 드라이버 초기화를 진행해줘야합니다. 

이 때는 ray.init()을 적어주시면 됩니다. 가끔 주피터로 하시다가 이미 실행중이라는 오류가 날 수도 있는데요. 

이런 경우는 ray.shutdown()을 이용해서 꺼준 뒤, ray.init()을 다시 실행시켜주면 됩니다. 

ray.shutdown()
ray.init()

 

이후 앞서 정의해준 함수를 호출시켜줍니다. 

저는 10000개의 입력을 for문으로 주었습니다. 

여기서 받아온 result는 Object들의 리스트라고 보시면 됩니다. 

result = [square.remote(i) for i in range(10000)]

 

실제로 출력하보면 아래와 같이 출력됩니다. 

[ObjectRef(c8ef45ccd0112571ffffffffffffffffffffffff0100000001000000), 
ObjectRef(16310a0f0a45af5cffffffffffffffffffffffff0100000001000000), 
ObjectRef(c2668a65bda616c1ffffffffffffffffffffffff0100000001000000), 
ObjectRef(32d950ec0ccf9d2affffffffffffffffffffffff0100000001000000), 
ObjectRef(e0dc174c83599034ffffffffffffffffffffffff0100000001000000), 
ObjectRef(f4402ec78d3a2607ffffffffffffffffffffffff0100000001000000), 
ObjectRef(f91b78d7db9a6593ffffffffffffffffffffffff0100000001000000), 
ObjectRef(82891771158d68c1ffffffffffffffffffffffff0100000001000000), 
ObjectRef(8849b62d89cb30f9ffffffffffffffffffffffff0100000001000000), 
ObjectRef(80e22aed7718a125ffffffffffffffffffffffff0100000001000000), 
ObjectRef(359ec6ce30d3ca2dffffffffffffffffffffffff0100000001000000), 
ObjectRef(1e8ff6d236132784ffffffffffffffffffffffff0100000001000000), 
ObjectRef(85748392bcd969ccffffffffffffffffffffffff0100000001000000), 
ObjectRef(d695f922effe6d99ffffffffffffffffffffffff0100000001000000), 
ObjectRef(2751d69548dba956ffffffffffffffffffffffff0100000001000000), 
ObjectRef(71b133a11e1c461cffffffffffffffffffffffff0100000001000000), 
ObjectRef(5d4b8d1788f12d2dffffffffffffffffffffffff0100000001000000), 
ObjectRef(c54e76759b2a0c10ffffffffffffffffffffffff0100000001000000), 
ObjectRef(239c2f70c73fbf73ffffffffffffffffffffffff0100000001000000), 
ObjectRef(1e360ffa862f8fe3ffffffffffffffffffffffff0100000001000000), 
...

 

이제 마지막으로 반환된 ObjectRef의 값을 뽑아봅시다. 

이 때는 ray.get()을 이용하시면 됩니다. 

ray.get(result)

 

위의 값을 출력해보면 아래와 같이 square함수의 출력 값입니다. 

[0, 1, 4, 9, 16, 25, 36, 49, 64, 81, 100, 121, 144, 169, 
196, 225, 256, 289, 324, 361, 400, 441, 484, 529, 576, 625, 
676, 729, 784, 841, 900, 961, ....]

 

정말 간단하죠?

전체 코드를 한번 보고 가실게요~!

import ray

@ray.remote
def square(x):
    return x**2

ray.shutdown()
ray.init()

result = [square.remote(i) for i in range(10000)]
print(ray.get(result))

 

정말 간단하죠!

 

사용 API 정리

자주 사용되는 API를 정리해보겠습니다.

API	설명	예시
ray.init()	ray를 사용하기 전 처음으로 초기화 해주는 함수	ray.init()
ray.shutdown()	ray의 프로세스를 종료시키는 함수	ray.shutdown()
@ray.remote	병렬처리할 함수를 정의해주는 데코레이터	@ray.remote def func(x):    ...
function.remote()	병렬처리 함수를 호출시키는 함수	func.remote(x)
ray.get()	병렬처리 완료된 값들을 가져오는 함수	result = ray.get(func.remote(x))
ray.put()	특정 값을 Object Store에 저장하는 함수	arr = np.random.random(100) arr = ray.put(arr)