具有文本或空间的多处理

Question

我正在尝试通过文本的并行化来加快大型文本列表的处理速度。当我从多处理中使用池时，生成的文本语料库是空的。我不确定问题是否在于我使用文本或多处理范式的方式？ 这是说明我的问题的示例：

import spacy
import textacy
from multiprocessing import Pool
texts_dict={
"key1":"First text 1."
,"key2":"Second text 2."
,"key3":"Third text 3."
,"key4":"Fourth text 4."
}
model=spacy.load('en_core_web_lg')
# this works
corpus = textacy.corpus.Corpus(lang=model)
corpus.add(tuple([value, {'key':key}],) for key,value in texts_dict.items())
print(corpus) # prints Corpus(4 docs, 8 tokens)
print([doc for doc in corpus])
# now the same thing with a worker pool returns empty corpus
corpus2 = textacy.corpus.Corpus(lang=model)
pool = Pool(processes=2) 
pool.map( corpus2.add, (tuple([value, {'key':key}],) for key,value in texts_dict.items()) )
print(corpus2) # prints Corpus(0 docs, 0 tokens)
print([doc for doc in corpus2])
# to make sure we get the right data into corpus.add
pool.map( print, (tuple([value, {'key':key}],) for key,value in texts_dict.items()) )

文本性基于空间性。Spacy不支持多线程，但应该可以在多个进程中运行。https://github.com/explosion/spaCy/issues/2075

根据@constt https://stackoverflow.com/a/58317741/4634344 的强烈建议，将结果收集到语料库中适用于大至n_docs= 10273 n_sentences= 302510 n_tokens= 2053129的数据集。

对于更大的数据集(16000 个文档 3MM 令牌(，我收到以下错误：

result_corpus=corpus.get() 
File "<string>", line 2, in get
File "/usr/lib/python3.6/multiprocessing/managers.py", line 772, in _callmethod
raise convert_to_error(kind, result)
multiprocessing.managers.RemoteError: 
---------------------------------------------------------------------------
Unserializable message: Traceback (most recent call last):
File "/usr/lib/python3.6/multiprocessing/managers.py", line 283, in serve_client
send(msg)
File "/usr/lib/python3.6/multiprocessing/connection.py", line 206, in send
self._send_bytes(_ForkingPickler.dumps(obj))
File "/usr/lib/python3.6/multiprocessing/connection.py", line 393, in _send_bytes
header = struct.pack("!i", n)
struct.error: 'i' format requires -2147483648 <= number <= 2147483647

' 我会调查，但如果您有直接的解决方案 - 非常感谢！

Answer 1

由于 python 进程在单独的内存空间中运行，因此您必须在池中的进程之间共享corpus对象。为此，您必须将corpus对象包装到一个可共享的类中，您将向 BaseManager 类注册该类。下面介绍了如何重构代码以使其正常工作：

#!/usr/bin/python3
from multiprocessing import Pool
from multiprocessing.managers import BaseManager
import spacy
import textacy

texts = {
'key1': 'First text 1.',
'key2': 'Second text 2.',
'key3': 'Third text 3.',
'key4': 'Fourth text 4.',
}

class PoolCorpus(object):
def __init__(self):
model = spacy.load('en_core_web_sm')
self.corpus = textacy.corpus.Corpus(lang=model)
def add(self, data):
self.corpus.add(data)
def get(self):
return self.corpus

BaseManager.register('PoolCorpus', PoolCorpus)

if __name__ == '__main__':
with BaseManager() as manager:
corpus = manager.PoolCorpus()
with Pool(processes=2) as pool:
pool.map(corpus.add, ((v, {'key': k}) for k, v in texts.items()))
print(corpus.get())

输出：

Corpus(4 docs, 16 tokens)