python - 如何使用 Python 对大 XML 文件执行查询？

Question

我有一个 7 GB 的 XML 文件，它是关于一家公司的所有交易，我只想过滤去年(2015 年)的记录。 一个文件的结构是:

<Customer>
<Name>A</Name>
<Year>2015<Year>
</Customer>

我还有它的 DTD 文件。 我不知道如何将这些数据过滤到文本文件中。 有没有这方面的教程或者库可以使用。

欢迎!

Answer 1

由于您的数据很大，我假设您已经决定无法将全部数据加载到内存中。这将是使用 DOM 样式(文档对象模型)解析器的方法。您实际上已经将您的问题标记为“SAX”(XML 的简单 API)，这进一步暗示您知道您需要一种非内存方法。

我想到了两种方法:

使用grep

有时对于 XML，使用纯文本处理工具会很有用。 grep 将允许您以纯文本形式过滤 XML 文档并找到 2015 的出现:

$ grep -B 2 -A 1 "<Year>2015</Year>"

-B 和 -A 选项指示 grep 打印匹配项周围的一些上下文行。

但是，只有当您的 XML 也是结构精确的纯文本时，这种方法才有效，而绝对没有必要(作为 XML)。也就是说，您的 XML 可以有任何空格的组合(或根本没有)并且在语义上仍然相同，但是 grep 方法取决于确切的空格排列。

萨克斯

因此，更可靠的非内存方法是使用 SAX。 SAX 实现在概念上非常简单，但编写起来有点乏味。本质上，您必须重写一个类，该类提供在源 XML 文档中发生某些“事件”时调用的方法。在xml.sax.handler标准库中的模块，这个类是ContentHandler。这些方法包括:

• 开始元素
• 结束元素
• 人物

然后，您覆盖的方法将决定如何处理这些事件。在 startElement(name, attrs) 的典型实现中，您可以测试 name 参数以确定元素的标签名称是什么。然后您可能会维护一堆您输入的元素。当 endElement(name) 发生时，您可能会从堆栈中弹出顶部元素，并可能对已完成的元素进行一些处理。 characters(content) 在源文档中遇到字符数据时发生。在此方法中，您可能会考虑构建一个字符数据字符串，然后在遇到 endElement 时可以对其进行处理。

因此对于您的特定任务，这样的事情可能会起作用:

from xml.sax import parse
from xml.sax.handler import ContentHandler

class filter2015(ContentHandler):
    def __init__(self):
        self.elements = []          # stack of elements
        self.char_data = u''        # string buffer
        self.current_customer = u'' # name of customer
        self.current_year = u''

    def startElement(self, name, attrs):
        if name == u'Name':
            self.elements.append(u'Name')
        if name == u'Year':
            self.elements.append(u'Year')

    def characters(self, chars):
        if len(self.elements) > 0 and self.elements[-1] in [u'Name', u'Year']:
            self.char_data += chars

    def endElement(self, name):
        self.elements.pop() if len(self.elements) > 0 else None

        if name == u'Name':
            self.current_customer = self.char_data
            self.char_data = ''
        if name == u'Year':
            self.current_year = self.char_data
            self.char_data = ''

        if name == 'Customer':
            # wait to check the year until the Customer is closed
            if self.current_year == u'2015':
                print 'Found:', self.current_customer

            # clear the buffers now that the Customer is finished
            self.current_year = u''
            self.current_customer = u''
            self.char_data = u''

source = open('test.xml')
parse(source, filter2015())