Builder.shin

Python 멀티프로세싱으로 CSV 대량 처리 성능 개선

문제 상황

매일 새벽 실행되는 배치 작업에서 고객 데이터 CSV(약 50만 건)를 파싱하고 변환하는 작업이 있었다. 단순 반복문으로 처리하니 약 40분이 소요됐고, 이후 작업들이 지연되는 문제가 발생했다.

기존 코드

import csv
import pandas as pd

def process_row(row):
    # 데이터 정제 및 변환 로직
    return transformed_row

with open('data.csv', 'r') as f:
    reader = csv.DictReader(f)
    results = [process_row(row) for row in reader]

단일 코어로 순차 처리하다 보니 CPU 사용률이 25% 정도에 머물렀다.

해결 방법

multiprocessing.Pool을 사용해 4개 프로세스로 분산 처리하도록 변경했다.

from multiprocessing import Pool
import csv

def process_row(row):
    # 동일한 변환 로직
    return transformed_row

def read_csv_chunk(filename):
    with open(filename, 'r') as f:
        reader = csv.DictReader(f)
        return list(reader)

if __name__ == '__main__':
    rows = read_csv_chunk('data.csv')
    
    with Pool(processes=4) as pool:
        results = pool.map(process_row, rows)
    
    # 결과 저장
    pd.DataFrame(results).to_csv('output.csv', index=False)

결과

  • 처리 시간: 40분 → 10분
  • CPU 사용률: 25% → 95%
  • 메모리 사용량은 약간 증가했지만 허용 범위 내

주의사항

  • if __name__ == '__main__': 가드는 필수다. 없으면 프로세스가 무한 생성된다.
  • 처리할 데이터가 피클 가능한(picklable) 객체여야 한다.
  • 메모리에 전체 데이터를 올리기 부담스러우면 청크 단위로 나눠서 처리하는 방식을 고려해야 한다.

간단한 변경으로 큰 효과를 봤다. I/O 바운드 작업이 아니라면 멀티프로세싱이 효과적이다.

Python 멀티프로세싱으로 CSV 대량 처리 성능 개선