解压文件的 Python 并行处理

Question

我是 python 中并行处理的新手。我在下面有一段代码，它遍历所有目录并解压缩所有 tar.gz 文件。然而，这需要相当多的时间。

import tarfile
import gzip
import os

def unziptar(path):
    for root, dirs, files in os.walk(path):
        for i in files:
            fullpath = os.path.join(root, i)
            if i.endswith("tar.gz"):
                print 'extracting... {}'.format(fullpath)
                tar = tarfile.open(fullpath, 'r:gz')
                tar.extractall(root)
                tar.close()

path = 'C://path_to_folder'
unziptar(path)

print 'tar.gz extraction completed'

我一直在浏览一些关于 multiprocessing 和 joblib 包的帖子，但我仍然不清楚如何修改我的脚本以并行运行。任何帮助表示赞赏。

编辑：@tdelaney

感谢您的帮助，令人惊讶的是，修改后的脚本解压缩所有内容所需的时间是原来的两倍（60 分钟与原始脚本的 30 分钟相比）！

我查看了任务管理器，看起来虽然使用了多核，但 CPU 使用率很低。我不确定为什么会这样。

Answer 1

创建一个池来完成这项工作非常容易。只需将提取器拉出到单独的工人中即可。

import tarfile
import gzip
import os
import multiprocessing as mp

def unziptar(fullpath):
    """worker unzips one file"""
    print 'extracting... {}'.format(fullpath)
    tar = tarfile.open(fullpath, 'r:gz')
    tar.extractall(os.path.dirname(fullpath))
    tar.close()

def fanout_unziptar(path):
    """create pool to extract all"""
    my_files = []
    for root, dirs, files in os.walk(path):
        for i in files:
            if i.endswith("tar.gz"):
                my_files.append(os.path.join(root, i))

    pool = mp.Pool(min(mp.cpu_count(), len(my_files))) # number of workers
    pool.map(unziptar, my_files, chunksize=1)
    pool.close()


if __name__=="__main__":
    path = 'C://path_to_folder'
    fanout_unziptar(path)
    print 'tar.gz extraction has completed'