使用多处理划分和征服etree.iterparse

Question

因此，让我们设想一个大的xml文档(文件大小>100mb)，我们想要使用cElementTree.iterparse进行迭代解析。

但英特尔向我们承诺的所有这些核心都是值得的，我们如何使用它们呢？这是我想要的：

from itertools import islice
from xml.etree import ElementTree as etree

tree_iter = etree.iterparse(open("large_file.xml", encoding="utf-8"))

first = islice(tree_iter, 0, 10000)
second = islice(tree_iter, 10000)

parse_first()
parse_second()

这似乎有几个问题，尤其是iterparse()返回的迭代器似乎拒绝切片。

有没有办法将大型xml文档的解析工作划分为两个或四个独立的任务(而不将整个文档加载到内存中？其目的是在单独的处理器上执行任务。

Answer 1

我认为你需要一个很好的线程池和一个任务队列来完成这个任务。我发现(并使用)了这个非常好的版本(它是用python3编写的，但是转换到2.x应该不会太难)：

# http://code.activestate.com/recipes/577187-python-thread-pool/

from queue import Queue
from threading import Thread

class Worker(Thread):
    def __init__(self, tasks):
        Thread.__init__(self)
        self.tasks = tasks
        self.daemon = True
        self.start()

    def run(self):
        while True:
            func, args, kargs = self.tasks.get()
            try: func(*args, **kargs)
            except Exception as exception: print(exception)
            self.tasks.task_done()

class ThreadPool:
    def __init__(self, num_threads):
        self.tasks = Queue(num_threads)
        for _ in range(num_threads): Worker(self.tasks)

    def add_task(self, func, *args, **kargs):
        self.tasks.put((func, args, kargs))

    def wait_completion(self):
        self.tasks.join()

现在，您只需在iterparse上运行循环，并让线程池为您分配工作。使用它很简单，如下所示：

def executetask(arg):
    print(arg)

workers = threadpool.ThreadPool(4) # 4 is the number of threads
for i in range(100): workers.add_task(executetask, i)

workers.wait_completion() # not needed, only if you need to be certain all work is done before continuing