在script from this answer的基础上,我有以下场景：一个包含2500个大文本文件的文件夹(每个约55Mb),所有制表符分隔. Web日志,基本上.

我需要md5哈希每个文件的每一行中的第二个’列’,将修改后的文件保存在别处.源文件位于机械磁盘上​​,目标文件位于SSD上.

该脚本非常快速地处理前25个(左右)文件.然后它减慢了WAY.基于前25个文件,它应该在2分钟左右完成所有文件.但是,根据之后的表现,完成它们需要15分钟(左右).

它运行在具有32 Gb RAM的服务器上,任务管理器很少显示超过6 Gb的使用情况.我已经设置了启动6个进程,但核心上的cpu使用率很低,很少超过15％.

为什么这放慢了？读/写磁盘问题？垃圾收集器？坏代码？有关如何加快速度的任何想法？

这是脚本

import os

import multiprocessing
from multiprocessing import Process
import threading
import hashlib

class ThreadRunner(threading.Thread):
    """ This class represents a single instance of a running thread"""
    def __init__(self,fileset,filedirectory):
        threading.Thread.__init__(self)
        self.files_to_process = fileset
        self.filedir          = filedirectory

    def run(self):
        for current_file in self.files_to_process:

            # Open the current file as read only
            active_file_name = self.filedir + "/" + current_file
            output_file_name = "D:/hashed_data/" + "hashed_" + current_file

            active_file = open(active_file_name,"r")
            output_file = open(output_file_name,"ab+")

            for line in active_file:
                # Load the line,hash the username,save the line
                lineList = line.split("\t")

                if not lineList[1] == "-":
                    lineList[1] = hashlib.md5(lineList[1]).hexdigest()

                lineOut = '\t'.join(lineList)
                output_file.write(lineOut)

            # Always close files after you open them
            active_file.close()
            output_file.close()

            print "\nCompleted " + current_file

class ProcessRunner:
    """ This class represents a single instance of a running process """
    def runp(self,pid,numThreads,filedirectory):
        mythreads = []
        for tid in range(numThreads):
            th = ThreadRunner(fileset,filedirectory)
            mythreads.append(th) 
        for i in mythreads:
            i.start()
        for i in mythreads:
            i.join()

class ParallelExtractor:    
    def runInParallel(self,numProcesses,filedirectory):
        myprocs = []
        prunner = ProcessRunner()

        # Store the file names from that directory in a list that we can iterate
        file_names = os.listdir(filedirectory)

        file_sets = []
        for i in range(numProcesses):
            file_sets.append([])

        for index,name in enumerate(file_names):
            num = index % numProcesses
            file_sets[num].append(name)


        for pid in range(numProcesses):
            pr = Process(target=prunner.runp,args=(pid,file_sets[pid],filedirectory)) 
            myprocs.append(pr) 
        for i in myprocs:
            i.start()

        for i in myprocs:
            i.join()

if __name__ == '__main__':    

    file_directory = "E:/original_data"

    processes = 6
    threads   = 1

    extractor = ParallelExtractor()
    extractor.runInParallel(numProcesses=processes,numThreads=threads,filedirectory=file_directory)