python多处理共享计数器，酸洗错误

Question

我正在处理一些非常大的文件。

我可以使用 collections.Counter 集合计算特定字符串的出现次数，该集合使用 multiprocessing.BaseManager 子类在进程之间共享。

虽然我可以共享计数器并且看似腌制，但似乎没有正确腌制。我可以将字典复制到可以腌制的新字典中。

我正在尝试了解如何在选择共享计数器之前避免“复制”它。

这是我的（伪代码）：

from multiprocessing.managers import BaseManager
from collections import Counter

class MyManager(BaseManager):
    pass

MyManager.register('Counter', Counter)

def main(glob_pattern):
    # function that processes files
    def worker_process(files_split_to_allow_naive_parallelization, mycounterdict):
        # code that loops through files
        for line in file:
            # code that processes line
            my_line_items = line.split()
            index_for_read = (my_line_items[0],my_line_items[6])
            mycounterdict.update((index_for_read,))

    manager = MyManager()
    manager.start()
    mycounterdict = manager.Counter()

    # code to get glob files , split them with unix shell split and then chunk then

    for i in range(NUM_PROCS):
        p = multiprocessing.Process(target=worker_process , args = (all_index_file_tuples[chunksize * i:chunksize * (i + 1)],mycounterdict))
        procs.append(p)
        p.start()
    # Now we "join" the processes
    for p in procs:
        p.join()

    # This is the part I have trouble with
    # This yields a pickled file that fails with an error
    pickle.dump(mycounterdict,open("Combined_count_gives_error.p","wb"))

    # This however works
    # How can I avoid doing it this way?
    mycopydict = Counter()
    mydictcopy.update(mycounterdict.items())
    pickle.dump(mycopydict,open("Combined_count_that_works.p","wb"))

当我尝试加载总是较小的固定大小的“腌制”错误文件时，我收到一个没有意义的错误。

如何在不通过上面伪代码中创建新字典的情况下腌制共享字典。

>>> p = pickle.load(open("Combined_count_gives_error.p"))
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/pickle.py", line 1378, in load
    return Unpickler(file).load()
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/pickle.py", line 858, in load
    dispatch[key](self)
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/pickle.py", line 880, in load_eof
    raise EOFError
EOFError

Answer 1

您的代码存在几个问题。首先，如果你让它悬空，你不能保证关闭文件。其次，mycounterdict 不是实际的Counter，而是它的代理——腌制它，你会遇到很多问题，因为它在这个过程之外是不可腌制的。但是，您也不需要使用 update 进行复制：.copy 会创建一个新的 Counter 副本。

因此你应该使用

with open("out.p", "wb") as f:
    pickle.dump(mycounterdict.copy(), f)

至于这是否是一个好的模式，答案是否。您应该在每个进程中单独计数，而不是使用共享计数器，以获得更简单的代码：

from multiprocessing import Pool
from collections import Counter
import pickle

def calculate(file):
    counts = Counter()
    ...
    return counts

pool = Pool(processes=NPROCS)
counts = Counter()
for result in pool.map(calculate, files):
    counts += result

with open("out.p", "wb") as f:
    pickle.dump(counts, f)