如何从大型 XML 文档中获取流式迭代器 [Node]？答案

【问题标题】：How to get a streaming Iterator[Node] from a large XML document?如何从大型 XML 文档中获取流式迭代器 [Node]？
【发布时间】：2012-01-21 11:29:10
【问题描述】：

我需要处理包含大量独立记录的 XML 文档，例如

<employees>
    <employee>
         <firstName>Kermit</firstName>
         <lastName>Frog</lastName>
         <role>Singer</role>
    </employee>
    <employee>
         <firstName>Oscar</firstName>
         <lastName>Grouch</lastName>
         <role>Garbageman</role>
    </employee>
    ...
</employees>

在某些情况下，这些只是大文件，但在其他情况下，它们可能来自流媒体源。

我不能只使用 scala.xml.XmlLoader.load() 它，因为我不想将整个文档保存在内存中（或等待输入流关闭），而我只需要使用一个一次记录。我知道我可以使用 XmlEventReader 将输入作为 XmlEvent 序列进行流式传输。然而，这些比 scala.xml.Node 更不方便使用。

所以我想以某种方式从中获得一个惰性 Iterator[Node]，以便使用方便的 Scala 语法对每个单独的记录进行操作，同时控制内存使用。

为此，我可以从 XmlEventReader 开始，在每个匹配的开始和结束标记之间建立一个事件缓冲区，然后从中构造一个节点树。但是，有没有我忽略的更简单的方法？感谢您提供任何见解！

【问题讨论】：

标签： xml scala xml-parsing

【解决方案1】：

您可以使用XMLEventReader 到ConstructingParser 使用的底层解析器，并通过回调处理顶层以下的员工节点。您只需小心处理后立即丢弃数据：

import scala.xml._

def processSource[T](input: Source)(f: NodeSeq => T) {
  new scala.xml.parsing.ConstructingParser(input, false) {
    nextch // initialize per documentation
    document // trigger parsing by requesting document

    var depth = 0 // track depth

    override def elemStart(pos: Int, pre: String, label: String,
        attrs: MetaData, scope: NamespaceBinding) {
      super.elemStart(pos, pre, label, attrs, scope)
      depth += 1
    }
    override def elemEnd(pos: Int, pre: String, label: String) {
      depth -= 1
      super.elemEnd(pos, pre, label)
    }
    override def elem(pos: Int, pre: String, label: String, attrs: MetaData,
        pscope: NamespaceBinding, nodes: NodeSeq): NodeSeq = {
      val node = super.elem(pos, pre, label, attrs, pscope, nodes)
      depth match {
        case 1 => <dummy/> // dummy final roll up
        case 2 => f(node); NodeSeq.Empty // process and discard employee nodes
        case _ => node // roll up other nodes
      }
    }
  }
}

然后您使用这样的方式来处理常量内存中第二级的每个节点（假设第二级的节点没有获得任意数量的子节点）：

processSource(src){ node =>
  // process here
  println(node)
}

与XMLEventReader 相比的好处是您不使用两个线程。此外，与您提出的解决方案相比，您不必两次解析节点。缺点是这依赖于ConstructingParser 的内部运作。

【讨论】：

太棒了！这很好用。从这种生成器风格的东西到迭代器并不难；看我的另一个答案。非常感谢！

【解决方案2】：

要从 huynhjl 的生成器解决方案获取 TraversableOnce[Node]，请使用 this trick：

def generatorToTraversable[T](func: (T => Unit) => Unit) = 
  new Traversable[T] {
    def foreach[X](f: T => X) {
      func(f(_))
    }
  }

def firstLevelNodes(input: Source): TraversableOnce[Node] =
  generatorToTraversable(processSource(input))

generatorToTraversable 的结果不能多次遍历（即使每次 foreach 调用都会实例化一个新的 ConstructingParser），因为输入流是一个 Source，它是一个 Iterator。但是，我们不能覆盖 Traversable.isTraversableAgain，因为它是最终的。

我们真的想通过返回一个迭代器来强制执行这一点。但是， Traversable.toIterator 和 Traversable.view.toIterator 都创建了一个中间流，它将缓存所有条目（违背了本练习的全部目的）。那好吧;如果流被访问两次，我会让流抛出异常。

还要注意整个事情不是线程安全的。

这段代码运行良好，我相信整体解决方案既懒惰又不缓存（因此是常量内存），尽管我还没有在大量输入上尝试过。

【讨论】：