[Python] 멀티 프로세싱 사용하기 - 멀티 프로세싱 적용을 위한 함수들 알아보자

목차

multiprocessing 사용하게된 계기

기록용으로  안읽고 다음 탭으로 넘어가셔도 됩니다.

파이썬에서 처리 속도를 높이기 위해 멀티 프로세싱을 사용할 수 있다.

대용량의 csv 파일 수 천개를 다뤄야 했던 경험이 있다.

pandas를 사용해서 파일을 읽어오는 것에만 상당히 많은 시간을 소모한다.

하나의 csv 파일을 읽어오기 위해서 그 동안 가만히 기다려야 하는 시간들이 매우 소모적이라고 생각했다.

대용량 csv 파일도 있지만 절반 정도는 매우 용량이 적은 csv 파일들이었다. 

메모리에 문제가 없는 수준에서는 여러개의 파일을 읽어와서 동시에 처리해도 괜찮겠다고 생각했다.

그래서 멀티프로세싱을 사용해 core 4개로 설정했더니 2배 이상 빨라졌다.  Good !

 

멀티 Pool 사용하기

from multiprocessing import Pool

if __name__ == '__main__':
    p = Pool(4)
    # do something here with Pool

    p.close() # or p.terminate()
    p.join()

 

종료하기 - close(), terminate()

생성한 Pool을 다 사용하였으면 적절한 위치에 close()과 join()을 호출해주는 것이 좋다.

join()은 Pool의 모든 프로세스들의 종료가 완료되기를 기다린다.

close()는 더 이상 Pool에 추가 작업이 들어가지 않는다는 것을 알려주며,

지금 수행 중인 작업이 모두 끝나면 Pool의 프로세스들을 종료한다.

terminate()를 사용하면,

현재 진행 중인 작업이 있더라도

즉시 Pool의 프로세스들을 종료한다.

 

apply()

Pool에게 작업 하나를 시킨다. 그리고 작업이 끝날 때까지 기다렸다가 결과를 받는다.

apply는 그 작업이 완료되지 않으면 메인 프로세스에서 다음 줄의 코드를 실행하지 않는다.

apply_async는 apply_async을 사용한 줄에서 작업이 다 끝나지 않아도 메인 프로세스의 다음 줄을 실행할 수 있다.

 

apply_async()

Pool에게 작업 하나를 시키고, AsyncResult를 반환받는다. 반환받은 AsyncResult에서 get()을 호출하면 작업의 반환 값을 얻을 수 있다.

단, 반환받은 AsyncResult의 get()을 호출한다면 그 작업이 끝나기 전까지는 메인 프로세스에서도 다음 줄로 넘어갈 수가 없다.

둘의 차이는 map(), map_async()와 starmap(), starmap_async()에서도 마찬가지이다.

from multiprocessing import Pool
import multiprocessing as mp
import time
import os

def func(num):
    c_proc = mp.current_process()
    print("Running on Process",c_proc.name,"PID",c_proc.pid)
    time.sleep(1)
    print("Ended",num,"Process",c_proc.name)
    return num

if __name__ == '__main__':
    p = Pool(4)
    start = time.time()

    ret1 = p.apply_async(func,(1,))
    ret2 = p.apply_async(func,(2,))
    ret3 = p.apply_async(func,(3,))
    ret4 = p.apply_async(func,(4,))
    ret5 = p.apply_async(func,(5,))
    print(ret1.get(),ret2.get(),ret3.get(),ret4.get(),ret5.get())

    delta_t = time.time()-start
    print("Time :",delta_t)

    p.close()
    p.join()

 

map() , map_async()

iterable에 대해 동일한 함수를 멀티프로세싱을 이용하여 처리하고자 할 때 사용한다. 단, 사용하고자 하는 함수는 단일 인자를 받아야 한다.

 

starmap() , starmap_async()

인자를 두 개 이상 받을 수 있다는 점을 제외하면 map(), map_async()와 같다.

 

imap(), imap_unordered()

map의 결과물은 list인 반면, imap의 결과물은 iterator이다.

기본 chunkzise는 1인데, 1 대신 적절하게 큰 값을 써주면 훨씬 빨리 처리할 수 있다.

결과물의 길이가 길어서 list로 나타내었을 때,

메모리에 부담이 가는 경우 imap을 사용해주면 좋다

imap_unordered()는 순서가 보장되지 않는다.

 

Callback

apply_async, map_async, starmap_async에서 callback 받기

위 세 개의 함수에 callback을 달아줄 수 있다.

callback의 인자는 위 함수들의 결과물을 받는다. 작업이 완료되는 순간 실행된다.

from multiprocessing import Pool
import multiprocessing as mp
import time
import os

def callback_func(result):
    print("callback_func got result :",result)

def square(x):
    return x*x

if __name__ == '__main__':
    with Pool(4) as p:
        result = p.map_async(square,range(11),callback=callback_func)
        result.wait()

callback_func got result : [0, 1, 4, 9, 16, 25, 36, 49, 64, 81, 100]

 

 

Reference

아래 블로그에 예시 코드가 너무 잘나와있다.

내용을 잊지 않기 위해 해당 게시글을 퍼온 글이다.

자세한 예시를 보고싶다면 아래 블로그 글을 참고하면 좋다.

https://tempdev.tistory.com/27?category=845382 

Python multiprocessing.Pool 멀티프로세싱 2

 

참고자료

멀티프로세싱에 tqdm을 사용해서 진행상황을 알고싶다면 아래 블로그를 참고하면 아주 이해가 쉽다. 검색해서 나오는 것 중 가장 잘 나와있는 자료라고 생각된다.

https://leimao.github.io/blog/Python-tqdm-Multiprocessing/

Progress Bars for Python Multiprocessing Tasks

]