使用多处理的Python代码并行化

Question

我对python多线程处理非常陌生，并且尝试并行化我的代码，因为它运行起来太长了。我有一个代码，它通过一大块数据来查找是否有任何文件损坏。到目前为止，我的代码是：

def check_Corrupt_1(dirPath, logfile):

    fileCheck = open(logfile, "w").close()
    fileCheck = open(logfile, "w")

    emptydir = []
    zero_size = {}
    #entering the year to be checked (day number)
    for fname in os.listdir(dirPath):

        if(os.listdir(os.path.join(dirPath, fname)) == []):
            emptydir.append(fname)

        else:

            #this makes sure that we do not enter an empty directory
            if fname not in emptydir:
                inPath = os.path.join(dirPath, fname)

                for filename in os.listdir(inPath):
                    hdfinfo = os.stat(os.path.join(inPath, filename))

                    if(hdfinfo.st_size == 0):
                        zero_size[filename] = True

                    else:

                        filepath = "/path/to/file"

                        strin = subprocess.Popen(["hdp", "dumpsds", "-h", os.path.join(inPath, filename)], stdout=subprocess.PIPE).communicate()[0]
                        #print(strin)
                        cmd = 'echo $?'
                        callno = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE)
                        #print(int(callno.stdout.read()[0]))
                        if(int(callno.stdout.read()[0]) != 0):
                            fileCheck.write(os.path.join(inPath, filename) + '\n')

我每年有365个目录，每个目录都包含很多要检查的文件。我正在运行bash命令来检查文件是否损坏，但是因为我正在运行的bash命令有一个很长的输出，所以这段代码要花很多时间来运行。我希望并行化将有助于使它更快，但不知道如何做到这一点。有没有别的方法比多重处理更快呢？如果有任何帮助，我将不胜感激。你知道吗

Answer 1

简单地浏览一下您的文章和您发布的代码段，大部分繁重的工作似乎都是通过hdp命令完成的。这就是你想要的parallelize。 您似乎正在打开一个子流程。 你也可以尝试使用线程。你的代码应该是这样的

#!/usr/bin/python
import thread
from subprocess import call

def check_Corrupt_1(dirPath, logfile):

    fileCheck = open(logfile, "w").close()
    fileCheck = open(logfile, "w")

    emptydir = []
    zero_size = {}
    #entering the year to be checked (day number)
    for fname in os.listdir(dirPath):

        if(os.listdir(os.path.join(dirPath, fname)) == []):
            emptydir.append(fname)

        else:

            #this makes sure that we do not enter an empty directory
            if fname not in emptydir:
                inPath = os.path.join(dirPath, fname)

                for filename in os.listdir(inPath):
                    hdfinfo = os.stat(os.path.join(inPath, filename))

                    if(hdfinfo.st_size == 0):
                        zero_size[filename] = True

                    else:
                        try:    
                            thread.start_new_thread(call(["hdp", "dumpsds", "-h"]))
                        except:
                            print "Error generating thread"

                        if(int(callno.stdout.read()[0]) != 0):
                            fileCheck.write(os.path.join(inPath, filename) + '\n')