python multiprocessing worker la consommation de mémoire augmente indéfiniment

Question

J'ai une liste de fichiers Excel 2010 (xlsx) qui doivent être recherchés pour une valeur spécifique. Comme xslx est un format binaire, cela ne peut pas être fait avec des éditeurs de texte normaux. Donc, je fais ce qui suit pour chaque fichier

1. obtenir le nom de fichier
2. ouvert pandas géants
3. convertir dataframe à numpy tableau
4. tableau de contrôle pour la valeur

Cela appelle à multitraitement comme n'est pas lié à I/O. Le temps est nécessaire pour les choses pandas et la conversion de tableau. J'ai donc mis en place une version multi-traitement de mon script (voir ci-dessous):

Le problème est la consommation de mémoire de chaque processus de travail. Il s'accumule continuellement en peaking dans 2GB par travailleur bien que chaque fichier xlsx soit simplement 100kb. Je ne vois pas pourquoi la mémoire n'est pas libérée avant qu'un nouveau fichier soit traité. De cette façon, je manque de mémoire avant que ma liste de fichiers ne soit traitée.

Le problème ne semble pas être la file d'attente, mais les pandas.

Voici mon code. Il peut être testé avec tous les fichiers xlsx que vous avez sur votre système.

import pandas as pd 
import multiprocessing as mp 
import glob 

path = r'c:\temp' 
fileFilter = 'serial.xlsx' 
searchString = '804.486' 


def searchFile(tasks, results, searchString): 
    """Iterates over files in tasks and searches in file for the 
    occurence of 'searchString'. 

    Args: 
    ----- 
    tasks: queue of strings 
     Files to look in 
    results: queue of strings 
     Files where the searchString was found 
    searchString: str 
     the string to be searched 
    """ 
    # for files in the queue 
    for task in iter(tasks.get, 'STOP'): 
     # read the filestructre into memory 
     xfile = pd.ExcelFile(task) 
     # iterate all sheets 
     for sheet in xfile.sheet_names[:3]: 
      # read the sheet 
      data = pd.read_excel(xfile, sheet) 
      # check if searchString is in numpy representation of dataframe 
      if searchString in data.values.astype(str): 
       # put filename in results queue 
       results.put(task) 
       break 
     xfile.close() 

if __name__ == "__main__": 
    # get all files matching the filter that are in the root path 
    print('gathering files') 
    files = glob.glob(path + '\**\{}'.format(fileFilter), recursive=True) 

    # setup of queues and variables 
    n_proc = 2 
    tasks = mp.Queue() 
    results = mp.Queue() 

    print('Start processing') 
    # setup processes and start them 
    procs = [mp.Process(target=searchFile, 
         args=(tasks, results, searchString)) 
      for x in range(n_proc)] 
    for p in procs: 
     p.daemon = True 
     p.start() 

    # populate queue 
    for file in files: 
     tasks.put(file) 

    for proc in procs: 
     tasks.put('STOP') 

    for p in procs: 
     p.join() 

    # print results 
    for result in range(results.qsize()): 
     print(results.get()) 

    print('Done') 

Answer 1

Semble problème dans gc, qui ne peut pas rassembler des cadres dans le contexte de pandas fonction que vous ne laissez jamais. Vous pouvez utiliser multiprocessing.Pool.map qui peut gérer la file d'attente pour vous. La fonction de travail sera appelée pour chaque article et laissez gc faire le travail. Vous pouvez également utiliser maxtasksperchild Param de constructeur de pool pour limiter le nombre d'éléments traités par le worker.

import glob 
import multiprocessing 


def searchFile(task, searchString): 
    xfile = pd.ExcelFile(task) 
    ... 
    if found: 
     return task 


if __name__ == '__main__': 
    files = glob.glob(path + '\**\{}'.format(fileFilter), recursive=True) 
    searchString = '804.486' 

    pool = multiprocessing.Pool(2, maxtasksperchild=10) 

    args = ((fname, searchString) for fname in files) 
    matchedFiles = filter(None, pool.map(searchFile, args)) 
    pool.close()