[Ray] Ray를 활용한 데이터 프레임 Concat

❗ Ray를 활용해서 데이터프레임을 분산/병렬처리 해보자.

 

대용량의 데이터를 다루다보면 데이터 처리하는 시간이 많이 걸릴 때가 있다. 

 

필자 역시 일을 하면서 경험했었고 실제로 “병렬로 데이터를 처리하면 훨씬 빠르게 처리 할 수 있지 않을까?” 라는 생각을 했던적이 있다.

특히, 가용가능한 CPU와 GPU가 있는데 이것들의 사용량을 보면 제대로 활용하지 못하고 있다는 생각이 들었다. 그래서 한정된 자원에서 최대한 효율을 내려고 병렬처리를 할 수 있는 Python 라이브러리를 찾아보는 계기가 되었다.

Python에서는 주로 Multiprocessing을 많이 사용하고 그 다음으로 Ray를 많이 사용하는 것으로 파악된다.

 

필자는 ML/DL환경에서도 쉽게 사용할 수 있는 Ray를 사용하기로 했다.

 

Ray는 python의 분산 컴퓨팅 프레임워크이고 적은 코드 수정으로 분산/병렬 처리가 가능하기때문에 쉽게 접근이 가능하다. 또한 ML/DL과 결합하여 생태계를 구축하고 있다. (ML/DL 학습, 모델링, 서빙까지 가능한 것으로 파악된다.)

자세한 것은 Ray를 방문하여 확인해보자

 

필자는 가끔 여러번 데이터 프레임을 Concatenation해야 할 때가 있다.

그럴때마다 순차적으로 pd.concat을 사용하는데 데이터의 양이 많으면 많은 시간이 걸린다.

이를 Ray를 통해 데이터를 병렬로 처리해 concat을 하는게 효과적이라고 생각했고 우리가 평소 concat 하는 방법(직렬)과 Ray를 활용하여 concat 하는 방법(병렬)의 속도차이를 실험해보았다.

 

먼저, 데이터를 만들어주겠다.

import pandas as pd
import numpy as np

data_array = np.array([[i for i in range(1000)] for j in range(10000)])
data_df = pd.DataFrame(data_array)

print(f"Data_df shape is {data_df.shape}")

# Data_df shape is (10000, 1000)

 

 data_array라는 변수에 행 10,000개 열 1,000개 데이터를 생성하여 데이터프레임으로 만든다. 0~999가 10,000개 줄이 있다고 생각하면 된다.

 

✔ Normal Concat 버전

   total_df라는 빈 데이터프레임을 만든다음 total_df에 data_df를 100번 concat해준다.

total_df = pd.DataFrame()

start_time = time.time()

for i in range(100):
    total_df = pd.concat([total_df,data_df])
    
end_time = time.time()
execution_time = end_time - start_time
print("Single : Execution Time:", execution_time, "seconds")
print(f"total_df shape {total_df.shape}")

# Single : Execution Time: 57.20992636680603 seconds
# total_df shape (1000000, 1000)

• 100번 Concat 결과 약 57초의 시간소요

 

✔ Ray Concat 버전

 

Ray를 사용하기전에 간단하게 Ray의 사용법을 알아보자.

• @ray.remote 
  →데코레이터로 함수에 사용하거나 클래스에 사용가능
• ray.init()
  → 여기에 num_cpus, num_gpus를 정의해주면 ray를 사용하는 모든 함수에 적용된다.
  → ray를 초기화하는 작업
• .remote()
  →@ray.remote와 연결되는 부분으로 데코레이터를 붙였다면 .remote()로 실행하면된다. 만약 함수 안에 인자가 들어가야한다면 .remote(인자) 이렇게 사용하면된다.
• ray.get()
  → 원래 값으로 돌려주는 역할을 한다. 이 함수를 실행하면 분산처리가 된다. 그래서 get은 마지막에 한번만 실행한다.       for문으로 여러번 실행하게 되면 오버헤드가 발생함. → 분산처리를 쓰는 이유가 없어짐.
• ray.shutdown()
  → ray를 종료한다.

1. 데코레이터로 Ray를 사용할 함수를 정의해준다.

import pandas as pd
import numpy as np
import ray
import time

ray.init(ignore_reinit_error=True,num_cpus=20, num_gpus = 1) # ray 초기화
total_df = pd.DataFrame()

@ray.remote
def concat_df(data_df):
    return data_df

** ray에서 ray.available_resources()를 통해서 가용가능한 gpu, cpu, memory 등을 알 수 있다. 필자는 cpu 20개와 gpu 1개를 사용하겠다.

 

2. (1) @ ray.remote를 감싼 함수인 concat_df 호출하여 for문으로 ray_list라는 리스트에 100번 apppend
        →  ray가 한번에 실행될 수 있는 리스트를 만든다.
    (2) ray.get()으로 lay_list에 있는 것들을 한번에 처리하게 됨
        → 하나의 데이터프레임으로 만들기 위해 df_new를 concat한다.

start_time = time.time()

ray_list = []
for i in range(100):
    ray_list.append(concat_df.remote(data_df))
    
    
df_new = ray.get(df)
df_new = pd.concat(df_new)

end_time = time.time()
execution_time = end_time - start_time
print("RAY : Execution Time:", execution_time, "seconds")
print(f"df shape {df_new.shape}")

# RAY  Execution Time: 2.5854716300964355 seconds
# df shape (1000000, 1000)

• ray_list안에 있는 ObjectRef 100개를 분산으로 처리하여 concat한 결과 약 2.5초 소요

---

✔ Normal Concat VS Ray Concat

각각 10번씩 실행한 그래프

 

Ray를 사용하면 약간의 편차가 있지만 대략 2~3초정도로 처리가 되고 Single로 작업하게 되면 57초의 시간이 소요된다. 단순 concat작업은 소요시간이 20배정도 차이가 나는데 어떤 작업을 하느냐에 따라 상이 할 것으로 보인다.

concat작업 뿐만 아니라 다양한 방면으로 사용이 가능 할 듯하다. 사용법을 잘 터득하면 가용할 수 있는 컴퓨팅 자원을 활용하여 조금더 효과적인 작업을 할 수 있을것이다.

 

ML/DL작업시 유용하게 사용 될 것으로 파악되며 관련 내용들은 추후에 기술하도록 하겠다.