python joblib으로 병렬 프로그래밍을 해보자

효율적인 프로그래밍을 위해 병렬 프로그래밍을 하곤 합니다. 특히 대용량의 데이터을 처리할 때 필수적이죠. joblib은 파이썬 프로그래밍에서 병렬처리를 가능하게 만들어줍니다. 파이썬에는 병렬 연산을 위한 디폴트 패키지로 multiprocessing이 있습니다. multiprocessing은 pandas의 DataFrame을 다루는 데에 유용하며, joblib은 일반적인 python 코딩에 두루두루 쓰입니다.

 

joblib의 dump, load 메소드는 python 객체를 저장하거나 불러오게끔 할 수 있기에, 머신러닝 프로젝트에서 모델을 저장하기에 유용하게 쓰이지만, 본 게시물에서는 병렬 처리를 도와주는 joblib의 Parallel, delayed에 대한 설명을 하겠습니다. 😁

 

설치

pip install joblib

joblib은 별도 설치가 필요합니다. pip를 통해 받아줍시다.

 

 

joblib.Parallel 사용하기

병렬 연산을 주지 않은 코드는 다음과 같습니다.

>>> from math import sqrt
>>> [sqrt(i ** 2) for i in range(10)]
#[0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]

 

joblib을 이용한 병렬연산 코드는 다음과 같습니다.

>>> from math import sqrt
>>> from joblib import Parallel, delayed
>>> Parallel(n_jobs=2)(delayed(sqrt)(i ** 2) for i in range(10))
[0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]

 

스레드 기반 병렬처리병렬 처리 vs 프로세스 기반 병렬 처리

joblib.Parallel은 기본값으로 'loky' 백엔드 모듈을 사용합니다. loky는 Python worker들이 동시에 다른 CPU를 사용하게 만들어줍니다(즉, multi-processing을 시행합니다.) 그러나 이 방법은 input, output 데이터를 직렬화해야하기 때문에 상당한 오버헤드 문제를 일으킬 수 있습니다.

오버헤드란? 프로그램의 실행 흐름 도중에 동떨어진 위치의 코드를 실행시켜야 할 때, 추가적으로 시간, 메모리, 자원이 사용되는 현상입니다. (출처: https://gamestory2.tistory.com/15)

 

내가 호출하는 함수들이 Python Global Interpreter Lock(GIL)을 해제하는 코드들로 이루어져 있다면? 멀티 프로세스보다는 멀티 스레드를 사용하는 것이 좋습니다. ex) Cython의 with nogil 구문으로 gil을 해제할 수 있습니다.

GIL(Global Interpreter Lock)이란? python 코드를 실행할 때에 여러 thread를 사용할 경우, 단 하나의 thread만이 python object에 접근할 수 있도록 제한하는 mutex입니다. 그 이유는 CPython이 메모리를 관리하는 방법이 thread-safe 하지 않기 때문입니다. (출처: https://dgkim5360.tistory.com/entry/understanding-the-global-interpreter-lock-of-cpython)

 

스레드를 사용해서 효율적으로 코드를 짜려면, joblib.Parallel의 파라미터로 prefer='threads'를 전달하면 됩니다. 그러면 자동으로 loky 백엔드에서 threading 백엔드를 사용합니다.

>>> Parallel(n_jobs=2, prefer="threads")(
...     delayed(sqrt)(i ** 2) for i in range(10))
[0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]

 

또는 수동으로 특정 백엔드를 설정할 수 있습니다.

>>> from joblib import parallel_backend
>>> with parallel_backend('threading', n_jobs=2):
...    Parallel()(delayed(sqrt)(i ** 2) for i in range(10))
[0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]

 

delayed의 역할

위 코드에서, Parallel의 입력값에 delayed를 붙여주는 이유는 뭘까요? document상에서 정의된 역할은 다음과 같습니다.

The delayed function is a simple trick to be able to create a tuple (function, args, kwargs) with a function-call syntax.

 

코드를 통해 알아봅시다.

Parallel(n_jobs=8)(getHog(i) for i in allImages)

이 코드는 다음 순서로 동작합니다.

1. Parallel 인스턴스를 8개의 worker로 동작하게 만든다.

2. 리스트 [getHog(i) for i in allImages]를 생성한다.

3. 생성한 리스트를 Parallel 인스턴스로 pass 한다.

 

여기서 문제가 뭘까요? 원래 목적은 allImages의 item들을 getHog에 전달하는 걸 병렬적으로 처리하는 것이었습니다. 그런데 2번 단계에서 이미 연산을 통해 목적이 달성돼버렸습니다. 8개의 worker를 사용하지 않고 말이죠.🤣 이러면 Parallel로 구현하는 의미가 없어져버립니다.

 

delayed는 함수와 거기에 들어갈 파라미터를 튜플로 생성합니다. 가령 foo(2, g=3)이라는 함수를 실행하고 싶을 때 delayed(foo)(2, g=3)은 (foo, [2], {g: 3})이라는 튜플을 생성해냅니다.

Parallel(n_jobs=8)(delayed(getHog)(i) for i in allImages)

delayed를 포함한 코드는 다음 순서로 동작합니다.

1. Parallel 인스턴스를 8개의 worker로 동작하게 만든다.

2. delayed를 통해 [(getHog, [img1], {}), (getHog, [img2], {}), ... ] 이라는 튜플 리스트를 생성한다.

3. Parallel의 튜플 리스트를 전달하고, 8개의 worker로 병렬 처리를 실행한다.

 

---

joblib의 Parallel 사용법에 대해 알아보았습니다. 많은 양의 데이터를 처리해야 하는 상황에서는 joblib을 이용해 병렬 처리하는 것이 시간을 단축시켜줍니다. 그러나 간단한 연산이나 특정 코드에서는 오히려 병렬 처리가 더 오래 걸리는 것에 유의하면서 사용합시다. 자세한 내용은 joblib document에 있습니다.

Joblib: running Python functions as pipeline jobs — joblib 1.2.0.dev0 documentation