Multiprocessing - Iteratively loop through files in folders

Question

I have to loop through 30 zip folders, and each zip folder has 50,000 - 90,000 jpeg files. Ideally, I would loop through each zip folder because unzipping each folder would take too long. For each file, I need to open each file, extract key information from it, and store the information into a list. Based on How to do multithreading on a folder with several files?, I tried enabling multiprocessing to make things quicker, however, I can't figure it out. In my example below, I am trying to get it to work with one folder at the moment, and then I will need to figure out how to make it loop through all 30 zip folders.

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)

What happens is my code just runs forever like it does without enabling multiprocessing. If I need to do multi-threading, I have six cores. Any advice would be appreciated.

Answer 1

You are missing several imports. But these major items I notice right away are:

1. for folder in os.listdir(): You are looping on every file and directory in the current directory yet the loop makes no reference to any of these files/directories; you are processing files101.zip repeatedly.
2. You appear to be running under Windows based on your chdir command. Code that creates new processes must within a if __name__ == '__main__': block.
3. Each process in a processing pool has its own address space and would be appending to different instances of data_list. Returning data_list back to the main process and having the main process appending all the return values to a master list would work, but would have to ensure that your image_processor function starts out with an empty data_list for every call.

I am assuming you want to process all the files with extension .zip in the current directory (of which there are approximately 30). I would modify your processing so that each submitted task to the processing pool's unit of work is not a single file within a zip archive (in which case you would be submitting 30 * 50,000 tasks), but rather a whole archive. So your main processing function is no longer image_processor but rather zip_processor. I have made a few other changes to this file, which make sense to me (I hope I did not break anything):

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)

Now since there is a lot of I/O involved, this might run very well using multithreading, in which case a larger pool size would be advantageous. Make the following changes:

#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.