I have a python script which exists of 2 processes:
- process 1: Loading and unzipping files
- process 2: Processing the files, doing some stuff with it.
Before implementing multiprocessing the software seemed to do its work in chronological order. Loading all the zipped files, unzipping them, then opening them to do some stuff with it.
So I have brought multiprocessing into the game and right now it seems that whilst the files are being loaded and unzipped the process of opening and doing stuff with them has already started. So there are multiple processes doing stuff at the same time. The problem is that when I run this code on big data (more then 100+ files) I get problems with concurrent file access. This results in PermissionError: [WinError 32] The process cannot access the file because it is being used by another process: When I run the snippet on a small data set (30 files approx.) it seems to go okay because the files are being unzipped really fast just in time process 2 starts.
What I want: I want to keep the multiprocessing because it is speeding up things but I want process 2 only to start if all the files have been unzipped (e.g. process 1 is done).
This is my snippet:
import os
import csv
import collections
import datetime
import zipfile
import re
import shutil
import fnmatch
from pathlib import Path
import ntpath
import configparser
from multiprocessing import Pool
def generate_file_lists():
# Change the following line to a real path
data_files = 'c:\desktop\DataEnergy'
pattern = '*.zip'
last_root = None
args = []
for root, dirs, files in os.walk(data_files):
for filename in fnmatch.filter(files, pattern):
if root != last_root:
last_root = root
if args:
yield args
args = []
args.append((root, filename))
if args:
yield args
def unzip(file_list):
"""
file_list is a list of (root, filename) tuples where
root is the same for all tuples.
"""
# Change the following line to a real path:
counter_part = 'c:\desktop\CounterPart'
for root, filename in file_list:
path = os.path.join(root, filename)
date_zipped_file_s = re.search('-(.\d+)-', filename).group(1)
date_zipped_file = datetime.datetime.strptime(date_zipped_file_s, '%Y%m%d').date()
#Create the new directory location
new_dir = os.path.normpath(os.path.join(os.path.relpath(path, start='c:\desktop\DataEnergy'), ".."))
#Join the directory names counter_part and create their paths.
new = os.path.join(counter_part, new_dir)
#Create the directories
if (not os.path.exists(new)):
os.makedirs(new)
zipfile.ZipFile(path).extractall(new)
#Get al the zipped files
files = os.listdir(new)
#Rename all the files in the created directories
for file in files:
filesplit = os.path.splitext(os.path.basename(file))
if not re.search(r'_\d{8}.', file):
os.rename(os.path.join(new, file), os.path.join(new, filesplit[0]+'_'+date_zipped_file_s+filesplit[1]))
# Required for Windows:
if __name__ == '__main__':
pool = Pool(13)
pool.map(unzip, generate_file_lists())
print('the files have been unzipped!')
#Start proces 2
all_missing_dates = ['20210701', '20210702']
missing_dates = [datetime.datetime.strptime(i, "%Y%m%d").date() for i in all_missing_dates]
dates_to_process = []
root = Path('.\middle_stage').resolve()
at_set = {'Audi', 'Mercedes', 'Volkswagen'}
#Only read the rows if they fulfill the following conditions.
def filter_row(r, missing_date):
if set(row).intersection(at_set):
if len(r) > 24 and r[24].isdigit():
aantal_pplgs = int(r[24])
date_time = datetime.datetime.fromisoformat(r[0]) if len(r) > 3 else True
condition_3 = date_time.date() == missing_date if len(r) > 3 else True
return condition_3
return False
#Open the files and read the rows
print("Start reading data")
data_per_date = dict()
for missing_date in missing_dates:
print("\tReading missing date: ", missing_date)
files=[fn for fn in (e for e in root.glob(f"**/*_{missing_date:%Y%m%d}.txt") if e.is_file())]
if len(files) != 13:
continue
dates_to_process.append(missing_date)
vehicle_loc_dict = collections.defaultdict(list)
for file in files:
with open(file, 'r') as log_file:
reader = csv.reader(log_file, delimiter = ',')
next(reader) # skip header
for row in reader:
if filter_row(row, missing_date):
print('filter_row has been executed!')
data_per_date[missing_date] = vehicle_loc_dict