멀티프로세싱/멀티스레딩 df 중복 관련 정리

이 말같지도 않은 코드를 기록하는 이유는 멍청한 실수를 방지하기 위함이다.

멀티~라는 것을 쓸 때에는 같은 자원을 참조하게끔 하면 예상치못한 결과가 발생하기 때문에

각자 자원을 쓸 수 있도록 장치가 필요하다.

 

---

멀티프로세싱으로 하면 df에 중복이 없고 멀티스레딩이나 그냥 for문으로 작동하면 중복이 생겼다.

중복제거를 하지않아도 중복이 없는 탓에 몇 시간 머리를 싸맸으나 테스트를 통해 이해를 하게되었다.

 

전역변수로 df 항목을 선언하였던게 문제였다.

멀티프로세싱으로 작업을 하게되면 같은 df를 쓰는게 아니라

프로세스 하나 마다 df를 새롭게 만들어서 사용하게 되는 것 같다.

따라서 이 df들을 concat해보면 중복이 없게된다.

아래는 테스트 코드이다.

import pandas as pd
from multiprocessing import Pool, freeze_support # Pool import하기

get_df = pd.DataFrame(columns=['순서'])

def test(number):
    print(f"----------{number}---------")
    for i in range(0,number):
        y = len(get_df)
        print(y)
        get_df.loc[y,"순서"]= i
    return get_df

if __name__ =="__main__":
    test_df=[0]
    freeze_support()
    get_list=[10,20,30]
    pool = Pool(processes = 4) 
    test_df[0] = pd.concat(pool.map(test,get_list))
    pool.close()                # 끝나면 닫아주고
    pool.join()                # 다른 프로세서들이 끝날 때까지 기다린다.

    print(test_df)

주의할 점은 빨리 끝나는 작업같은 경우에는 같은 df를 쓰게된다. 예시코드에서 숫자를 6, 7, 8 이런식으로 하면 같이쓴다.

즉 하나의 프로세스가 실행되고, 끝나기 전에 다른 프로세스가 실행된다면 중복되지않는다. 아마 보통의 작업은 이런식으로 될 것이다.

 

반면에 멀티임에도 하나의 프로세스가 끝난 뒤에 다른 프로세스가 실행되면 같이쓴다.

근본적으로는 df를 전역변수로 쓰지않아야한다는 것을 강조하고싶다.

 

멀티스레드는 중복 조차도 뒤죽박죽인 경우가 많다. 섬세하게 써야한다.

import asyncio
import requests
import time
import os
import threading
from concurrent.futures import ThreadPoolExecutor
import pandas as pd


get_df = pd.DataFrame(columns=['순서'])

def fetcher(url):

    i = f"{os.getpid()} process | {threading.get_ident()} url : {url}"
    y = len(get_df)
    get_df.loc[y,"순서"]= y
    return get_df

def main():
    urls=["https://www.naver.com","https://www.google.com","https://instagram.com"] * 50

    executor = ThreadPoolExecutor(max_workers=12)
    print(executor)

    params = [url for url in urls]
    test_df = pd.concat(list(executor.map(fetcher,params)))
    
    time.sleep(2)
    test_df.to_excel("123.xlsx",encoding="utf-8-sig",index=False)
        # print(result)

if __name__ == "__main__":
    # asyncio.run(main())
    start = time.time()
    main()
    end = time.time()
    print(end-start)

같은 코드임에도 이런 결과값이 나온다.

물론 당연한게 같은 자원(get_df)를 끌어쓰기 때문이긴하다만.

---

어느 영역에서도 똑같지만, 실력이 그렇게 높지않다면 데이터를 100%신뢰할 수 있는가에 대해서는 생각을 해봐야한다.

 

아래 코드처럼 df를 각 함수마다 구현한다면 일정한 결과값이 나온다.

def fetcher(url):
    get_df = pd.DataFrame(columns=['순서'])
    i = f"{os.getpid()} process | {threading.get_ident()} url : {url}"
    y = len(get_df)
    get_df.loc[y,"순서"]= y
    return get_df

이는 완벽한 해결책은 아니다.

이로인해서 처리해야 할 일이 더 늘 수도 있는데, 처리 속도보다 데이터의 신뢰도가 중요하다면(당연한 얘기지만) 추천한다.