Избегайте загрузки данных spaCy в каждом подпроцессе при многопроцессорной обработке

Question

Я хотел бы использовать spaCy в программе, которая в настоящее время реализована с многопроцессорной обработкой. В частности, я использую ProcessingPool для создания 4 подпроцессов, которые затем уходят и выполняют свои весёлые задачи.

Чтобы использовать spaCy (специально для тегов POS), мне нужно вызвать spacy.load('en'), что является дорогостоящим вызовом (занимает ~ 10 секунд). Если я загружаю этот объект в каждый подпроцесс, он займет ~ 40 секунд, так как все они читаются из того же места. Это досадно долго.

Но я не могу понять, как заставить их делиться объектом, который загружается. Этот объект не может быть маринованные, что означает (насколько я знаю):

1. Он не может быть передан в Pool.map вызова
2. Он не может храниться и использоваться Manager например, чтобы затем быть распределены между процессами

Что я могу сделать?

Answer 1

Я не понимаю, как вы используете Pool.map, но помните, что Pool.map не работает с массивным количеством ввода. В Python 3.6 он реализован в Lib/multiprocessing/pool.py, как вы можете видеть, он утверждает, что в качестве первого аргумента он принимает iterable, но реализация выполняет consume the whole iterable перед запуском многопроцессорной карты. Поэтому я думаю, что это не Pool.map, что вам нужно использовать, если вам нужно обработать много данных. Может быть Pool.imap и Pool.imap_unordered может работать.

О вашей реальной проблеме. У меня есть решение, которое не связано с Pool.map и работает вроде как multiprocess foreach.

Прежде всего, необходимо, чтобы наследовать Pool и создать рабочий процесс:

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. 

Вы подготовить пул отростка так:

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() 

Затем вы можете кормить бассейн через queue:

for args in iterable: 
    queue.put(args) 

iterable - это список аргументов arg которые вы передаете рабочим. Вышеупомянутый код будет толкать содержимое iterable как можно быстрее. В принципе, если работник будет достаточно медленным, почти все итерационные будут перенесены в очередь до того, как рабочие закончат свою работу. Вот почему содержимое итерации должно вписываться в память.

Если рабочие параметры (ака iterable.) Не может поместиться в памяти необходимо синхронизировать как-то главный отростка и рабочих ...

В конце убедитесь, назвать следующие:

for worker in workers: 
    queue.put(None) 

for worker in workers: 
    worker.join()