[Python] 파일에서 필요한 데이터를 추출해 파일로 저장하기 #3

오늘은 파이썬의 mmap 모듈과 multiprocessing 모듈을 이용해서 텍스트 파일의 데이터를 추출해 재가공하는 부분을 작성해봤다. 대용량 파일의 처리는 mmap 모듈을 사용하는 것이 효율적이라고 해서 적용을 했는데.. 일단 10만건 기준으로 기존보다는 10초정도 빨라진다. 추출 데이터를 좀 더 늘려서 테스트를 진행해봐야 할 것 같다.

# _*_ coding: utf-8 _*_

from multiprocessing import Process
import mmap
import time
import datetime
import locale

seller = []
exclude = []

def get_exce_time(start, end):
    return datetime.timedelta(seconds=(end - start))

def number_format(num, places=0):
    """Format a number according to locality and given places"""
    locale.setlocale(locale.LC_ALL, "")
    return locale.format("%.*f", (places, num), True)

def file_read():
    start = time.time()

    results = {}

    with open('seller_id.txt', 'r+b') as f:
        mm = mmap.mmap(f.fileno(), 0)

        for line in iter(f.readline, b''):
            str = line.split(b"\t")
            id = str[4].strip()

            if id not in seller:
                seller.append(id)

        mm.close()

        results['seller'] = seller

    with open('exclude_item_id.txt', 'r+b') as f:
        mm = mmap.mmap(f.fileno(), 0)

        for line in iter(f.readline, b''):
            str = line.strip()

            if str not in exclude:
                exclude.append(str)

        mm.close()

        results['exclude'] = exclude

    end = time.time()
    print("File Reading Time : {0}".format(get_exce_time(start, end)))

    return results

def find_item_code(data, process, idx, limit):
    seller = data['seller']
    exclude = data['exclude']

    for k, s_id in enumerate(seller):
        start = time.time()
        item = []

        if k % process != idx:
            continue

        with open('AllDataBasedOnDB.dat', 'r+b') as f:
            mm = mmap.mmap(f.fileno(), 0)

            for line in iter(mm.readline, b''):
                if limit > 0 and len(item) == limit:
                    break

                if line.find(s_id) == -1:
                    continue

                str = line.split(b"\t")
                code = str[0].strip()

                if code not in exclude and code not in item:
                    item.append(code)

            # txt 파일생성
            with open('./out/' + s_id.decode() + '.txt', 'w+b') as out:
                out.write(b'\n'.join(item))

            end = time.time()
            print("{0} Process #{1} Running Time : {2} Count: {3}".format(s_id.decode(), idx, get_exce_time(start, end),
                                                                          number_format(len(item))))

            mm.close()

if __name__ == '__main__':
    data = file_read()

    # 실행 프로세스 수
    process = 4

    # 추출한 레코드 수 0으로 설정하면 모든 레코드 추출
    limit = 0

    procs = [Process(target=find_item_code, args=((data, process, i, limit,))) for i in range(process)]
    for p in procs: p.start()

mmap 모듈도 그렇고 multiprocessing 모듈도 그렇고.. 파이썬을 접한지가 오래 되지 않아서 코드를 맞게 작성했는지 모르겠다.