Jak uzyskać Iterator strumieniowy [Węzeł] z dużego dokumentu XML?

Question

Potrzebuję przetwarzać dokumenty XML składające się z bardzo dużej liczby niezależnych rekordów, np.

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

W niektórych przypadkach są to po prostu duże pliki, ale w innych mogą pochodzić ze źródła strumieniowego.

Nie mogę po prostu scala.xml.XmlLoader.load(), ponieważ nie chcę przechowywać całego dokumentu w pamięci (lub czekać na zamknięcie strumienia wejściowego), gdy potrzebuję tylko pracować z jeden rekord na raz. Wiem, że mogę użyć XmlEventReader do streamowania danych wejściowych jako sekwencji XmlEvents. Są one jednak znacznie mniej wygodne w pracy niż scala.xml.Node.

Więc chciałbym jakoś wydostać z tego leniwego Iteratora [Węzeł], aby operować na każdym indywidualnym zapisie przy użyciu wygodnej składni Scala, jednocześnie kontrolując wykorzystanie pamięci.

Aby to zrobić samodzielnie, mógłbym zacząć od XmlEventReader, zbudować bufor zdarzeń pomiędzy każdym pasującym tagiem początkowym i końcowym, a następnie skonstruować z niego drzewo węzłów. Ale czy istnieje łatwiejszy sposób, który przeoczyłem? Dzięki za wszelkie spostrzeżenia!

Answer 1

Możesz używać podstawowego parsera używanego przez XMLEventReader do ConstructingParser i przetwarzać węzły pracowników poniżej najwyższego poziomu za pomocą wywołania zwrotnego. Po prostu trzeba być ostrożnym, odrzucając dane przetwarzane tak szybko, jak:

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 
     } 
    } 
    } 
} 

Następnie użyć tak aby przetworzyć każdy węzeł na drugim poziomie w pamięci stałej (zakładając, że węzły na drugim poziomie nie są coraz dowolna liczba dzieci):

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

zaletą w porównaniu do XMLEventReader jest to, że nie należy używać dwóch wątków. Ponadto nie trzeba analizować węzła dwa razy w porównaniu do proponowanego rozwiązania. Wadą jest to, że zależy to od wewnętrznego działania ConstructingParser.

Answer 2

Aby dostać się z generatora roztworem huynhjl jest do TraversableOnce[Node] użyć 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)) 

wynikiem generatorToTraversable nie jest przesuwny więcej niż raz (choć nowa ConstructingParser jest tworzony na każdym wywołaniu foreach) ze względu na wejście strumień to Źródło, które jest Iteratorem. Nie możemy jednak przesłonić Traversable.isTraversableAgain, ponieważ jest ostateczna.

Naprawdę chcielibyśmy to wymusić, zwracając po prostu Iterator. Jednak zarówno Traversable.toIterator, jak i Traversalable.view.toIterator tworzą strumień pośredni, który buforuje wszystkie wpisy (pokonując cały cel tego ćwiczenia). No cóż; Pozwolę, by strumień wyrzucił wyjątek, jeśli jest dostępny dwukrotnie.

Należy również pamiętać, że całość nie jest bezpieczna dla wątków.

Ten kod działa świetnie, i uważam, że ogólne rozwiązanie jest zarówno leniwym, jak i nie buforującym (stąd stała pamięć), chociaż nie próbowałem go jeszcze na dużym wejściu.