多处理 - 迭代循环浏览文件夹中的文件

Question

我必须遍历 30 个 zip 文件夹，每个 zip 文件夹有 50,000 - 90,000 个 jpeg 文件。理想情况下，我会遍历每个 zip 文件夹，因为解压缩每个文件夹会花费太长时间。对于每个文件，我需要打开每个文件，从中提取关键信息，并将信息存储到一个列表中。基于How to do multithreading on a folder with several files?，我尝试启用多处理以使事情变得更快，但是，我无法弄清楚。在下面的示例中，我目前正试图让它与一个文件夹一起工作，然后我需要弄清楚如何让它遍历所有 30 个 zip 文件夹。

import os
from zipfile import ZipFile

data_list = []
 
def image_processor(file):
    with ZipFile("files101.zip") as zip_file:
        with zip_file.open(file, "r") as img_file:
            img_data = img_file.readlines(1) # data is available in beginning of each file
            
            # Extract data #1
            pattern_1 = r'IMG:\d{,3}'
            if re.findall(pattern_1, str(img_data)):
                img_extract = re.findall(pattern_1, str(img_data))[0]
            else:
                img_extract = np.nan

            # Extract timestamp
            time_pattern = r'Time:\s\d{2}-\d{2}-\d{4}\s\s\d{2}:\d{2}:\d{2}'
            if re.findall(time_pattern, str(img_data)):
                time_extract = re.findall(time_pattern, str(img_data))[0]
            else:
                time_extract = np.nan

            # Create list   
            return data_list.append([img_extract, time_extract])

os.chdir(r"C:\\Users\\xxxxxx\\Desktop\\zip")
for folder in os.listdir():
    file_list = ZipFile("files101.zip", "r").namelist()

    with ProcessPool(processes=8) as pool:
        pool.map(image_processor, file_list)

发生的情况是我的代码永远运行，就像它没有启用多处理一样。如果我需要做多线程，我有六个核心。任何建议将不胜感激。

Answer 1

您缺少几个导入。但我马上注意到的这些主要项目是：

1. for folder in os.listdir(): 您正在循环当前目录中的每个文件和目录，但循环没有引用这些文件/目录中的任何一个；您正在重复处理 files101.zip。
2. 根据您的chdir 命令，您似乎正在Windows 下运行。创建新进程的代码必须在 if __name__ == '__main__': 块内。
3. 处理池中的每个进程都有自己的地址空间，并且会附加到data_list 的不同实例。将data_list 返回到主进程并让主进程将所有返回值附加到主列表将起作用，但必须确保您的image_processor 函数以空data_list 开始每个调用。

我假设您要处理当前目录中所有扩展名为 .zip 的文件（其中大约有 30 个）。我会修改您的处理，以便每个提交到处理池工作单元的任务不是 zip 存档中的单个文件（在这种情况下，您将提交 30 * 50,000 个任务），而是整个存档。所以你的主要处理函数不再是image_processor，而是zip_processor。我对该文件进行了一些其他更改，这对我来说很有意义（我希望我没有破坏任何东西）：

import os
import glob
from zipfile import ZipFile
import re
import numpy as np
from multiprocessing import Pool

def zip_processor(zipfile):
    with ZipFile(zipfile) as zip_file:
        data_list = []
        for file in zip_file.namelist():
            with zip_file.open(file, "r") as img_file:
                # take element 0 from returned list and convert to a string
                img_data = img_file.readlines(1)[0].decode() # data is available in beginning of each file
                # Extract data #1
                pattern_1 = r'IMG:\d{,3}'
                # why do findall if you are only using the first occurrence?
                m = re.search(pattern1, img_data)
                img_extract = m.group(0) if m else np.nan
    
                # Extract timestamp
                time_pattern = r'Time:\s\d{2}-\d{2}-\d{4}\s\s\d{2}:\d{2}:\d{2}'
                m = re.search(time_pattern, img_data)
                time_extract = m.group(0) if m else np.nan
    
                # Create list   
                data_list.append([img_extract, time_extract])
        return data_list
            

# required for Windows:
if __name__ == '__main__':
    os.chdir(r"C:\\Users\\xxxxxx\\Desktop\\zip")

    # Default pool size:
    with Pool() as pool:
        results = pool.imap(zip_processor, glob.iglob('*.zip'))
        data_list = []
        for result in results:
            data_list.extend(result)

现在由于涉及大量 I/O，因此使用多线程可能会运行得很好，在这种情况下，更大的池大小将是有利的。进行以下更改：

#from multiprocessing import Pool
from multiprocessing.pool import ThreadPool
... # etc.

if __name__ == '__main__':
    os.chdir(r"C:\\Users\\xxxxxx\\Desktop\\zip")

    zip_list = glob.glob('*.zip')
    with ThreadPool(len(zip_list)) as pool:
        results = pool.imap(zip_processor, zip_list)
        ... # etc.