用于解析大文件的多处理和共享多处理管理器列表

Question

我正在尝试使用下面的代码解析一个巨大的文件（大约 23 MB），其中我使用从文件中读取的所有行填充 multiprocessing.manager.list。在每个进程的目标例程 (parse_line) 中，我弹出一行并对其进行解析以创建具有某些解析属性的 defaultdict 对象，最后将这些对象中的每一个推送到另一个 multiprocessing.manager.list 中。

class parser(object):
def __init__(self):
    self.manager = mp.Manager()
    self.in_list = self.manager.list()
    self.out_list = self.manager.list()
    self.dict_list,self.lines, self.pcap_text = [],[],[]
    self.last_timestamp = [[(999999,0)]*32]*2
    self.num = Word(nums)
    self.word = Word(alphas)
    self.open_brace = Suppress(Literal("["))
    self.close_brace = Suppress(Literal("]"))
    self.colon = Literal(":")
    self.stime = Combine(OneOrMore(self.num + self.colon) + self.num + Literal(".") + self.num)
    self.date = OneOrMore(self.word) + self.num + self.stime
    self.is_cavium = self.open_brace + (Suppress(self.word)) + self.close_brace
    self.oct_id = self.open_brace + Suppress(self.word) + Suppress(Literal("=")) \
            + self.num + self.close_brace
    self.core_id = self.open_brace + Suppress(self.word) + Suppress(Literal("#")) \
            + self.num + self.close_brace
    self.ppm_id = self.open_brace + self.num + self.close_brace
    self.oct_ts = self.open_brace + self.num + self.close_brace
    self.dump = Suppress(Word(hexnums) + Literal(":")) + OneOrMore(Word(hexnums))
    self.opening = Suppress(self.date) + Optional(self.is_cavium.setResultsName("cavium")) \
            + self.oct_id.setResultsName("octeon").setParseAction(lambda toks:int(toks[0])) \
            + self.core_id.setResultsName("core").setParseAction(lambda toks:int(toks[0])) \
            + Optional(self.ppm_id.setResultsName("ppm").setParseAction(lambda toks:int(toks[0])) \
            + self.oct_ts.setResultsName("timestamp").setParseAction(lambda toks:int(toks[0]))) \
            + Optional(self.dump.setResultsName("pcap"))

def parse_file(self, filepath):
    self.filepath = filepath
    with open(self.filepath,'r') as f:
        self.lines = f.readlines()
        for lineno,line in enumerate(self.lines):
            self.in_list.append((lineno,line))
        processes = [mp.Process(target=self.parse_line) for i in range(mp.cpu_count())]
        [process.start() for process in processes]
        [process.join() for process in processes]

    while self.in_list:
        (lineno, len) = self.in_list.pop()
        print mp.current_process().name, "start"
        dic = defaultdict(int)
        result = self.opening.parseString(line)
        self.pcap_text.append("".join(result.pcap))
        if result.timestamp or result.ppm:
            dic['oct'], dic['core'], dic['ppm'], dic['timestamp'] = result[0:4]
            self.last_timestamp[result.octeon][result.core] = (result.ppm,result.timestamp)
        else:
            dic['oct'], dic['core'] = result[0:2]
            dic['ppm'] = (self.last_timestamp[result.octeon][result.core])[0]
            dic['ts'] = (self.last_timestamp[result.octeon][result.core])[1]
        dic['line'] = lineno
        self.out_list.append(dic)

但是，整个过程大约需要 3 分钟才能完成。

我的问题是，是否有更好的方法来加快速度？

我正在使用 pyparsing 模块来解析每一行，如果有什么不同的话。

PS：对 Paul McGuire 的建议进行了更改

Answer 1

不是一个大的性能问题，但要学会直接迭代文件，而不是使用 readlines()。代替此代码：

    self.lines = f.readlines()
    for lineno,line in enumerate(self.lines):
        self.in_list.append((lineno,line))

你可以写：

    self.in_list = list(enumerate(f))

一个隐藏的性能杀手正在使用while self.in_list: (lineno,line) = list.pop()。每次调用 pop 都会从列表中删除第 0 个元素。不幸的是，Python 的列表是作为数组实现的。要删除第 0 个元素，第 1..n-1 个元素必须在数组中向上移动一个插槽。您不必真的随时销毁self.in_list，只需对其进行迭代：

for lineno, line in self.in_list:
    <Do something with line and line no. Parse each line and push into out_list>

如果您认为使用 self.in_list 是一种节省内存的措施，那么您可以通过使用双端队列（来自 Python 提供的集合模块）来避免 Python 列表的数组转换效率低下。双端队列在内部以链表的形式实现，因此从任一端推送或弹出非常快，但索引访问很慢。要使用双端队列，请替换以下行：

    self.in_list = list(enumerate(f))

与：

    self.in_list = deque(enumerate(f))

然后将代码中的调用 self.in_list.pop() 替换为 self.in_list.popleft()。

但更可能是性能问题是您用于处理每一行的 pyparsing 代码。但是由于您没有发布解析器代码，因此我们无法提供太多帮助。

要了解时间的去向，请尝试保留所有代码，然后注释掉 <Do something with line and line no. Parse each line and push into out_list> 代码（您可能需要为 for 循环添加 pass 语句），然后运行你的 23MB 文件。这将使您大致了解 3 分钟中有多少时间用于读取和迭代文件，以及用于实际解析的时间。当您发现真正的性能问题出在哪里时，然后在另一个问题中回复。