用于 80+GB XML 的 Python sax 到 lxml

Posted 2023-02-16

【中文标题】用于 80+GB XML 的 Python sax 到 lxml

【英文标题】：Python sax to lxml for 80+GB XML

【发布时间】：2012-04-06 06:11:40

【问题描述】：

如何使用 sax 读取 XML 文件并将其转换为 lxml etree.iterparse 元素？

为了提供对该问题的概述，我使用 lxml 构建了一个 XML 摄取工具，用于 XML 提要，其大小范围为 25 - 500MB，需要每天两次摄取，但需要执行一次60 - 100GB 文件的时间摄取。

我选择使用 lxml 是基于详细说明节点大小不超过 4 -8 GB 的规范，我认为这将允许将节点读入内存并在完成时清除。

如果代码如下所示的概述

elements = etree.iterparse(
    self._source, events = ('end',)
)
for event, element in elements:
    finished = True
    if element.tag == 'Artist-Types':
        self.artist_types(element)

def artist_types(self, element):
    """
    Imports artist types

    :param list element: etree.Element
    :returns boolean:
    """
    self._log.info("Importing Artist types")
    count = 0
    for child in element:
        failed = False
        fields = self._getElementFields(child, (
            ('id', 'Id'),
            ('type_code', 'Type-Code'),
            ('created_date', 'Created-Date')
        ))
        if self._type is IMPORT_INC and has_artist_type(fields['id']):
            if update_artist_type(fields['id'], fields['type_code']):
                count = count + 1
            else:
                failed = True
        else:
            if create_artist_type(fields['type_code'],
                fields['created_date'], fields['id']):
                count = count + 1
            else:
                failed = True
        if failed:
            self._log.error("Failed to import artist type %s %s" %
                (fields['id'], fields['type_code'])
            )
    self._log.info("Imported %d Artist Types Records" % count)
    self._artist_type_count = count
    self._cleanup(element)
    del element

如果我可以添加任何类型的说明，请告诉我。

【问题讨论】：

那么问题是什么？您收到错误消息了吗？

问题就在第一句话中……为什么要投反对票？

你的问题有点奇怪。你为什么要使用 SAX？ iterparse 是SAX 的替代品。您可以从 SAX 事件生成 iterparse 事件，但为什么有人会这样做呢？

据我了解，lxml 不会流式传输文件并将其完全读入内存（或至少是正在读取的节点）。要流式传输它，我需要使用 SAX，但我已经在 lxml 中构建了整个摄取，并且转换是不可能的。

iterparse 不会将整个文件读入内存。它构建了一棵树，但是是增量的。使用clear()

处理完节点后删除节点

【参考方案1】：

iterparse 是一个迭代解析器。它将发出Element 对象和事件，并在解析时逐步构建整个Element 树，因此最终它将整个树都保存在内存中。

但是，有限制的内存行为很容易：在解析时删除不再需要的元素。

典型的“巨型 xml”工作负载是具有大量表示记录的子元素的单个根元素。我认为这是您正在使用的那种 XML 结构？

通常使用clear() 清空您正在处理的元素就足够了。您的内存使用量会增加一点，但不是很多。如果你有一个非常大的文件，那么即使是空的 Element 对象也会消耗太多，在这种情况下，你还必须删除以前看到的 Element 对象。请注意，您不能安全地删除当前元素。 lxml.etree.iterparse documentation describes this technique。

在这种情况下，每次找到</record> 时，您将处理一条记录，然后您将删除所有以前的记录元素。

下面是一个使用无限长 XML 文档的示例。它将在解析时打印进程的内存使用情况。请注意，内存使用量是稳定的，不会继续增长。

from lxml import etree
import resource

class InfiniteXML(object):

    def __init__(self):
        self._root = True

    def read(self, len=None):
        if self._root:
            self._root = False
            return "<?xml version='1.0' encoding='US-ASCII'?><records>\n"
        else:
            return """<record>\n\t<ancestor attribute="value">text value</ancestor>\n</record>\n"""

def parse(fp):
    context = etree.iterparse(fp, events=('end',))
    for action, elem in context:
        if elem.tag == 'record':
            # processing goes here
            pass
        
        # memory usage
        print resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
        
        # cleanup
        # first empty children from current element
            # This is not absolutely necessary if you are also deleting siblings,
            # but it will allow you to free memory earlier.
        elem.clear()
        # second, delete previous siblings (records)
        while elem.getprevious() is not None:
            del elem.getparent()[0]
        # make sure you have no references to Element objects outside the loop

parse(InfiniteXML())

【讨论】：

没有单一的“根”节点，而是将数据分解为 20 多个“根”节点，每个节点都包含自己的子集。当前工具的工作方式与您的代码有点相似，即在处理后删除任何不需要的节点，这允许处理相当大的数据块，但是一旦我尝试处理较大的节点之一“我假设大于 8GB 的​​大小”该进程将分段（在 for 循环中）for action, elem in context: 这让我相信它正在被读入内存。

您能展示一些示例 XML 吗？您发布的代码似乎仅显示一种主要元素类型。 Iterparse 不会将整个文件读入内存，因此只需将您的工作流程划分为较小的子树，这些子树确实适合内存，并在每次迭代后删除所有内容。

不幸的是，上面发布的代码大约是我所能提供的，但是在重写了大部分摄取后，导入现在可以使用上述方法工作。代码gist.github.com/2161849见下面sn-p。

【参考方案2】：

我在http://effbot.org/zone/element-iterparse.htm 找到了这个有用的示例。大胆强调是我的。

增量解析#

请注意，iterparse 仍然会构建树，就像 parse 一样，但您可以在解析时安全地重新排列或删除树的某些部分。例如，要解析大文件，您可以在处理完元素后立即删除它们：

for event, elem in iterparse(source):
    if elem.tag == "record":
        ... process record elements ...
        elem.clear()

上述模式有一个缺点；它不会清除根元素，因此您最终会得到一个包含许多空子元素的元素。 如果您的文件很大，而不仅仅是很大，这可能是个问题。要解决此问题，您需要掌握根元素。 最简单的方法是启用开始事件，并将对第一个元素的引用保存在变量中：

# get an iterable 
context = iterparse(source, events=("start", "end"))

# turn it into an iterator 
context = iter(context)

# get the root element 
event, root = context.next()

for event, elem in context:
    if event == "end" and elem.tag == "record":
        ... process record elements ...
        root.clear()

（未来版本将更容易从循环内访问根元素）

【讨论】：

感谢您的回答，但我已经对此进行了探索，至少从我的测试来看，节点仍然被完全读入内存并且没有流式传输

【参考方案3】：

这已经有几年了，我没有足够的声誉来直接评论已接受的答案，但我尝试使用它来解析一个 OSM，在那里我可以找到一个国家/地区的所有交叉点。我最初的问题是我的 RAM 用完了，所以我认为我必须使用 SAX 解析器，但找到了这个答案。奇怪的是，它没有正确解析，并且使用建议的清理以某种方式在读取 elem 节点之前清除它（仍然不确定这是如何发生的）。从代码中删除了elem.clear()，现在它运行得非常好！