Pandas DataFrame을 병렬처리 하는 방법

Scikit-learn의 모델들은 cython과 joblib으로 최적화 및 자동 병렬처리 되도록 설계되어 있지만, Pandas는 여전히 내부적으로 병렬처리 기능을 지원하지 않습니다.

하지만, 큰 규모의 DataFrame을 돌리다보면 전처리에도 시간이 많이 걸리게 됩니다. 그런 경우에 병렬처리를 통해 속도를 개선할 수 있습니다.

이 포스팅에서는 가장 간단한 CPU 프로세스 병렬처리를 다루도록 하겠습니다. 방법은 간단합니다. 거대한 DataFrame을 CPU 코어 수 만큼 분할하고, 전처리 기능을 수행한 다음 다시 합치면 됩니다.

import pandas as pd
import numpy as np
import seaborn as sns
from multiprocessing import Pool

num_cores = 4
iris = pd.DataFrame(sns.load_dataset('iris'))

예시로 iris 데이터를 사용하겠습니다. cpu 코어의 수는 multiprocessing.cpu_count() 함수를 통해서 얻으실 수 있습니다.

def parallelize_dataframe(df, func):
    df_split = np.array_split(df, num_cores)
    pool = Pool(num_cores)
    df = pd.concat(pool.map(func, df_split))
    pool.close()
    pool.join()
    return df

parallelize_dataframe은 어떤 전처리 함수가 들어왔을 때 CPU 병렬처리를 도와주는 함수입니다. multiprocessing.Pool을 이용하여 분할된 DataFrame에 함수를 적용시키고, pd.concat()으로 다시 합치는 과정입니다.

def multiply_columns(data):
    data['length_of_word'] = data['species'].apply(lambda x: len(x))
    return data

각 종 이름의 글자 수를 세는 전처리 함수를 예로 들어 속도차이를 확인해보겠습니다. 결과는 아래와 같습니다.

다른 방법으로 Pandas의 engine에 Dask를 사용하는 방법도 있습니다. http://dask.readthedocs.io/en/latest/