[python 심화] 14. Future 동시성/ 비동기 작업 실행/ ThreadPoolExecutor, ProcessPoolExecutor

코드 - 패캠 수업 코드 참고 (패캠 수업 정리)

 

 

<이전 글>

https://silvercoding.tistory.com/30

[python 심화] 13. 코루틴 (coroutine) , yield from

 

* Future 동시성 / 비동기 작업 실행 

: 지연시간 (Block) CPU 및 리소스 낭비 방지 -> (FILE) Network I/O관련 작업 동시성 활용 권장 

: 적합한 작업일 경우 순차 진행보다 압도적으로 성능 향상 

 

 

 

* 이번 포스팅에서 실습할 3가지 

1. 순차 실행 

2. concurrent.futures 방법1 

3. concurrent.futures 방법 2 

 

import os 
import time # 성능측정 
import sys
import csv

# 국가 정보 
NATION_LS = ('Singapore Germany Israel Norway Italy Canada France Spain Maxico').split()
# 초기 CSV 위치 
TARGET_CSV = './resources/nations.csv'
# 저장 폴더 위치 
DEST_DIR = './csvs'
# CSV 헤더 기초 정보 
HEADER = ['Region','Country','Item Type','Sales Channel','Order Priority','Order Date','Order ID','Ship Date','Units Sold','Unit Price','Unit Cost','Total Revenue','Total Cost','Total Profit']

<csv 파일 : 패스트 캠퍼스 제공>

NATION_LS : 사용할 국가 이름 리스트 

TARGET_CSV : csv파일이 있는 위치 경로 

DEST_DIR : 최종적으로 csv파일이 저장될 폴더 위치 경로 

HEADER : csv의 헤더를 추출 해놓은 리스트 

 

 

 

- 1. 순차실행 예제 

포스팅에서는 차례대로 함수를 작성하지만 공부할 때는 뒤에서부터( (6) 부터 ) 거슬러오며 공부하기

 

 

(1) 국가별 CSV 파일 저장 

def save_csv(data, filename):
    # 최종 경로 생성 
    path = os.path.join(DEST_DIR, filename)

    with open(path, 'w', newline="") as fp:
        writer = csv.DictWriter(fp, fieldnames=HEADER)
        # Header Write
        writer.writeheader()
        # Dict to CSV write
        for row in data: 
            writer.writerow(row)

최종 csv파일을 저장하는 함수이다. 국가별로 만들어진 데이터(data)와 생성할 파일이름(filename)을 인자로 받는다. 

 

 

(2) 국가별 분리 

def get_sales_data(nt):
    with open(TARGET_CSV, 'r') as f:
        reader = csv.DictReader(f)
        # Dict를 리스트로 적재 
        data = []
        # Header 확인 
        # print(reader.fieldnames)
        for r in reader:
            # OrderedDict 확인 
            # print(r)
            # 조건에 맞는 국가만 삽입 
            if r['Country'] == nt:
                data.append(r)
    return data

국가의 이름(nt)을 인자로 받는다. csv파일을 열어서 DictReader로 reader에 받아준다. Country가 nt국가일 때 data리스트에 r을 넣어준다.  이렇게 국가별로 데이터를 분리해 주는 함수이다. 

 

 

(3) 중간 상황 출력

def show(text):
    print(text, end=' ')
    # 중간 출력 (버퍼 비우기)
    sys.stdout.flush()

국가의 이름을 받아서 터미널에 출력해 주어 진행상황을 알려주는 함수이다. 

 

 

(4) 국가 별 분리 함수 실행 - (1) , (2) , (3) 함수 사용

def separate_many(nt_list):
    for nt in sorted(nt_list):
        # 분리 데이터 
        data = get_sales_data(nt)
        # 상황 출력 
        show(nt)
        # 파일 저장 
        save_csv(data, nt.lower() + '.csv')

    return len(nt_list)

여기서 for문으로 각 국가별 이름을 앞에서 보았던 get_sales_data에 넣어주어 분리된 데이터를 csv파일로 저장해주는 함수를 만든다. show함수로 진행상황을 출력하고, save_csv함수로 각 국가별 csv파일 데이터를 저장한다. 

 

 

(5) 시간 측정 및 메인함수 - (4) 함수 사용

def main(separate_many):
    # 시작 시간 
    start_tm = time.time()
    # 결과 건수 
    result_cnt = separate_many(NATION_LS)
    # 종료 시간 
    end_tm = time.time() - start_tm 

    msg = '\n{} csv separated in {:.2f}s'

    # 최종 결과 출력 
    print(msg.format(result_cnt, end_tm))

드디어 main함수에서는 separate_many함수를 사용하여 국가별 csv파일을 생성하도록 하고 , 이에 대한 시간을 측정하여 출력해준다. 

 

 

(6) 실행 - (5) 함수 사용

if __name__ == '__main__':
    main(separate_many)

Canada France Germany Israel Italy Maxico Norway Singapore Spain
9 csv separated in 21.09s

이렇게 main함수를 사용해 주면 앞에서 만들어 놓았던 함수들이 모두 실행되며 

국가별 csv 파일이 모두 만들어 진다. 

 

 

 

 

 

 

 

 

 

 

 

 

*** Google Python GIL (Global Interpreter Lock) 

: GIL은 한 번에 하나의 스레드만 수행할 수 있게 인터프리터 자체에서 락을 거는 것이다. 

 

<python GIL 관련문서>

https://wiki.python.org/moin/GlobalInterpreterLock

GlobalInterpreterLock - Python Wiki

 

 

- 2. concurrent.futures 방법1 (ThreadPoolExecutor, ProcessPoolExecutor)

<concurent.futures 참고문서>

https://docs.python.org/ko/3.7/library/concurrent.futures.html

concurrent.futures — 병렬 작업 실행하기 — Python 3.7.11 문서

: map() 사용

: 서로 다른 스레드 또는 프로세스에서 실행 가능 

: 내부 과정을 알 필요가 없으며 , 고수준으로 인터페이스를 제공함 . 

...
from concurrent import futures

concurrent의 futres 를 추가로 import 해준다. 

 

 

* (1) - (3) 의 함수는 앞과 동일하다. 

 

(1) 국가별 CSV 파일 저장 

def save_csv(data, filename):
    # 최종 경로 생성 
    path = os.path.join(DEST_DIR, filename)

    with open(path, 'w', newline="") as fp:
        writer = csv.DictWriter(fp, fieldnames=HEADER)
        # Header Write
        writer.writeheader()
        # Dict to CSV write
        for row in data: 
            writer.writerow(row)

(2) 국가별 분리

def get_sales_data(nt):
    with open(TARGET_CSV, 'r') as f:
        reader = csv.DictReader(f)
        # Dict를 리스트로 적재 
        data = []
        # Header 확인 
        # print(reader.fieldnames)
        for r in reader:
            # OrderedDict 확인 
            # print(r)
            # 조건에 맞는 국가만 삽입 
            if r['Country'] == nt:
                data.append(r)
    return data

(3) 중간 상황 출력

def show(text):
    print(text, end=' ')
    # 중간 출력 (버퍼 비우기)
    sys.stdout.flush()

 

 

(4) 국가 별 분리 함수 실행 - (1) , (2) , (3) 함수 사용

def separate_many(nt):
    # 분리 데이터 
    data = get_sales_data(nt)
    # 상황 출력 
    show(nt)
    # 파일 저장 
    save_csv(data, nt.lower() + '.csv')

    return nt

1번 방법인 순차실행에서는 for문을 이용하여 국가별 csv파일을 만들어냈다. 2번 방법에서는 map함수를 사용하여 국가 이름 리스트를 받지 않고 , 국가이름 하나씩을 받아 csv파일로 저장한다.  최종적으로 국가이름 하나(nt)를 리턴하도록 만든다.

 

 

 

(5) 시간 측정 및 메인 함수

futures.ThreadPoolExecutor() / futures.ProcessPoolExecutor 

 

 - ThreadPoolExecutor : GIL 종속 

def main(separate_many):
    # worker 개수
    worker = min(20, len(NATION_LS))
    # 시작 시간
    start_tm = time.time()
    # 결과 건수
    with futures.ThreadPoolExecutor(worker) as excutor:
        # map -> 작업 순서 유지, 즉시 실행
        result = excutor.map(separate_many, sorted(NATION_LS))
    # 종료 시간
    end_tm = time.time() - start_tm

    msg = '\n{} csv separated in {:.2f}s'
    # 최종 결과 출력
    print(msg.format(len(list(result)), end_tm))
    
    
# 실행
if __name__ == '__main__':
    main(separate_many)

 France Norway Spain Germany Maxico Canada Singapore Israel Italy 
 9 csv separated in 21.61s 

worker 라는 일꾼 변수를 만들어 주고, ThreadPoolExecutor()의 인자에 넣어준다. 여기서 result는 제너레이터 객체로 저장된다. 따라서 최종결과 출력에서 리스트의 길이를 알기 위해 result를 list 타입으로 바꿔주어야 한다. 

 

만약 그냥 len(result)로 작성한다면 !

 File "c:/Users/./Desktop/python_high/chapter06_03_02.py", line 100, in main  
    result_cnt = len(excutor.map(separate_many, sorted(NATION_LS))) 
 TypeError: object of type 'generator' has no len() 

제너레이터는 len() 를 갖지 않는다는 예외가 발생한다. 

 

 

 

 

- ProcessPoolExecutor : GIL 우회, 변경 후 -> 내부적으로 os.cpu_count() 메소드 호출 (운영체제 자체에서 우회에서 감 -> 속도가 빠름 )

def main(separate_many):
    # worker 개수
    worker = min(20, len(NATION_LS))
    # 시작 시간
    start_tm = time.time()
    # 결과 건수
    with futures.ProcessPoolExecutor() as excutor:
        # map -> 작업 순서 유지, 즉시 실행
        result = excutor.map(separate_many, sorted(NATION_LS))
    # 종료 시간
    end_tm = time.time() - start_tm

    msg = '\n{} csv separated in {:.2f}s'
    # 최종 결과 출력
    print(msg.format(len(list(result)), end_tm))
    
  
# 실행
if __name__ == '__main__':
    main(separate_many)

 Norway Canada Germany France Spain Singapore Italy Maxico Israel 
 9 csv separated in 8.11s 

같은 방법으로 ProcessPoolExecutor를 해본다. 시간이 훨씬 단축된 것을 볼 수 있다. 

 

언제 ProcessPoolExecutor쓰고, ThreadPoolExecutor쓰고, 코루틴을 쓰는지를 생각해야 한다. ProcessPoolExecutor쓰면 cpu 사용량이 급격히 증가하고 빠른 결과가 나온다. 그래서 이러한 멀티 프로세싱은 단기간의 딥러닝 고속 연산에 사용하는 것이 좋다. ThreadPoolExecutor사용하면 속도는 더 걸리지만 cpu 사용량은 잔잔하게 증가한다. 이 두가지의 차이를 구분하고 , 적절히 사용해야 한다. 

 

 

 

 

 

 

 

 

- 3. concurrent.futures 방법2 (map()함수 대신 for문을 이용하여 세부적인 과정 찍어보기) 

 

* (1) - (3) 의 함수는 앞과 동일하다. 

 

(4) 국가 별 분리 함수 실행 - (1) , (2) , (3) 함수 사용

def main(separate_many):
    # worker 개수
    worker = min(20, len(NATION_LS))
    # 시작 시간
    start_tm = time.time()
    # futures 
    futures_list = []
    # 결과 건수
    # ProcessPoolExecutor : GIL 우회, 변경 후 -> 내부적으로 os.cpu_count() 메소드 호출 (운영체제 자체에서 우회에서 감)
    # ThreadPoolExecutor : GIL 종속 
    # with futures.ThreadPoolExecutor(worker) as excutor:
    with futures.ProcessPoolExecutor() as excutor:
        # submit -> callable 객체 스케쥴링(실행 예약) -> future 
        # Future -> as_complete(), result(), done(), cancelled() 주로 사용 
        for nt in sorted(NATION_LS):
            # future 반환
            future = excutor.submit(separate_many, nt)
            # 스케쥴링
            futures_list.append(future)
            # 출력 
            # print('Scheduled for {} : {}'.format(nt, future))
            # print()

        for future in futures.as_completed(futures_list):
            result = future.result()
            done = future.done()
            cancelled = future.cancelled()
            # furue 결과 확인 
            print('Future Result : {}, Done : {}'.format(result, done))
            print('Future Cancelled : {}'.format(cancelled))

    # 종료 시간
    end_tm = time.time() - start_tm

    msg = '\n{} csv separated in {:.2f}s'
    # 최종 결과 출력
    print(msg.format(len(futures_list), end_tm))

for문으로 작업을 일일이 하고, 세부적으로 진행상황과 어떻게 마무리 되었는지를 알고 싶을 때 이 방법을 쓴다. 

2번 방법에서는 map()함수를 이용하여 separate_many 함수로 국가별 csv파일을 저장해 주었다. 3번 방법에서는 submit()함수를 이용하여 futre변수에 할당해주고 , 이 future를 futures_list에 append 해준다. 

 

futures.as_complted()의 인자에 futures_list를 넣어주고, 각각의 future에 대한 result, done, cancelled 값을 얻을 수 있다. 

 

 

* 실행 

if __name__ == '__main__':
    main(separate_many)

 Germany Canada Future Result : Germany, Done : True 
 Future Cancelled : False 
 Norway Israel Future Result : Norway, Done : True 
 Italy Future Cancelled : False 
 Future Result : Canada, Done : True 
 Future Cancelled : False 
 Future Result : Israel, Done : True 
 Future Cancelled : False 
 Future Result : Italy, Done : True 
 Future Cancelled : False 
 Singapore France Future Result : Singapore, Done : True 
 Future Cancelled : False 
 Future Result : France, Done : True 
 Future Cancelled : False 
 Maxico Future Result : Maxico, Done : True 
 Future Cancelled : False 
 Spain Future Result : Spain, Done : True 
 Future Cancelled : False 

 9 csv separated in 8.60s 

실행을 해보면 , result는 각 국가의 이름(nt : separate_many의 리턴값), Done은 모두 True, Cancelled는 모두 False가 나와 정상적으로 작동한 것을 볼 수 있다. 이와 같이 for문과 submit함수를 사용하여 세부적인 내용을 확인할 수도 있다.