Éviter de charger des données spaCy dans chaque sous-processus lors du multitraitement

Question

Je souhaite utiliser spaCy dans un programme qui est actuellement implémenté avec le multitraitement. Plus précisément, j'utilise ProcessingPool pour engendrer 4 sous-processus qui partent ensuite et font leurs tâches joyeuses.

Pour utiliser spaCy (spécifiquement pour l'étiquetage POS), je dois appeler spacy.load('en'), ce qui est un appel coûteux (prend ~ 10 secondes). Si je dois charger cet objet dans chaque sous-processus, cela prendra ~ 40 secondes, car ils lisent tous à partir du même endroit. C'est énormément long.

Mais je n'arrive pas à trouver un moyen de les amener à partager l'objet en cours de chargement. Cet objet ne peut pas être décapée, ce qui signifie (pour autant que je sache):

1. Il ne peut pas être passé dans le Pool.map appel
2. Il ne peut pas être stocké et utilisé par une instance Manager pour ensuite être partagé entre les processus

Que puis-je faire?

Answer 1

Je ne sais pas exactement comment utiliser Pool.map mais sachez que Pool.map ne fonctionne pas avec un grand nombre d'entrées. Dans Python 3.6, il est implémenté dans Lib/multiprocessing/pool.py comme vous pouvez le voir, il indique qu'il prend un iterable comme premier argument mais l'implémentation fait consume the whole iterable avant d'exécuter la carte multi-processus. Donc, je pense que ce n'est pas Pool.map que vous devez utiliser si vous avez besoin de traiter beaucoup de données. Peut-être que Pool.imap et Pool.imap_unordered peuvent fonctionner.

À propos de votre numéro actuel. J'ai une solution qui n'implique pas Pool.map et fonctionne un peu comme multiprocess foreach.

Vous devez d'abord hériter Pool et créer un processus de travail:

from multiprocessing import cpu_count 
from multiprocessing import Queue 
from multiprocessing import Process 


class Worker(Process): 

    english = spacy.load('en') 

    def __init__(self, queue): 
     super(Worker, self).__init__() 
     self.queue = queue 

    def run(self): 
     for args in iter(self.queue.get, None): 
      # process args here, you can use self. 

Vous préparez la piscine de Processus comme ça:

queue = Queue() 
workers = list() 
for _ in range(cpu_count()): # minus one if the main processus is CPU intensive 
    worker = Worker(queue) 
    workers.append(worker) 
    worker.start() 

Ensuite, vous pouvez nourrir la piscine via queue:

for args in iterable: 
    queue.put(args) 

iterable est la liste des arg que vous transmettez aux travailleurs. Le code ci-dessus va pousser le contenu de iterable aussi vite que possible. Fondamentalement, si le travailleur est assez lent, presque tous les itératifs seront poussés dans la file d'attente avant que les travailleurs aient terminé leur travail. C'est pourquoi le contenu de l'itérable doit entrer dans la mémoire.

Si les arguments des travailleurs (aka iterable.) Ne peuvent pas entrer dans la mémoire, vous devez synchroniser en quelque sorte le principal et les travailleurs Processus ...

A la fin assurez-vous d'appeler les éléments suivants:

for worker in workers: 
    queue.put(None) 

for worker in workers: 
    worker.join()