xml - Alpakka：如何将 xml 反序列化为对象 - 反序列化器有更简洁的模式吗？

Question

我有复杂的 XML 数据（它可能包含大量数据并且可能超过 15GB），这些数据具有复杂的性质和深层结构。我们需要对庞大的 XML 进行流处理。使用新的Alpakka库是我们的首选，因为它是一个很有前途的解决方案。

scala-xml 序列化和其他 Scala 库中存在过时的线程，但我们需要将大量 XML 作为事件流进行处理。

为了简化事情，让我们假设我们有一个PurchaseOrder（XML 来自这个页面）。

<?xml version="1.0"?>  
<PurchaseOrder PurchaseOrderNumber="99503" OrderDate="1999-10-20">  
  <Address Type="Shipping">  
    <Name>Ellen Adams</Name>  
    <Street>123 Maple Street</Street>  
    <City>Mill Valley</City>  
    <State>CA</State>  
    <Zip>10999</Zip>  
    <Country>USA</Country>  
  </Address>  
  <Address Type="Billing">  
    <Name>Tai Yee</Name>  
    <Street>8 Oak Avenue</Street>  
    <City>Old Town</City>  
    <State>PA</State>  
    <Zip>95819</Zip>  
    <Country>USA</Country>  
  </Address>  
  <DeliveryNotes>Please leave packages in shed by driveway.</DeliveryNotes>  
  <Items>  
    <Item PartNumber="872-AA">  
      <ProductName>Lawnmower</ProductName>  
      <Quantity>1</Quantity>  
      <USPrice>148.95</USPrice>  
      <Comment>Confirm this is electric</Comment>  
    </Item>  
    <Item PartNumber="926-AA">  
      <ProductName>Baby Monitor</ProductName>  
      <Quantity>2</Quantity>  
      <USPrice>39.98</USPrice>  
      <ShipDate>1999-05-21</ShipDate>  
    </Item>  
  </Items>  
</PurchaseOrder>

我正在尝试从 XML流式传输所有Item并反序列化它们。请注意，相同的标签可能会出现在不同的级别上。此外， Item中的元素/属性可以以任意顺序出现。我看到的一种方法（主要基于Alpakka 的 XmlProcessingTest - 有人可以提出更好的参考吗？），可能如下所示：

import akka.actor.ActorSystem
import akka.stream.alpakka.xml.scaladsl.XmlParsing
import akka.stream.alpakka.xml.{Characters, EndElement, ParseEvent, StartElement}
import akka.stream.scaladsl._
import akka.stream.{ActorMaterializer, IOResult}
import scala.collection.mutable
import scala.concurrent.duration._
import scala.concurrent.{Await, Future}
import akka.util.ByteString

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.alpakka.xml.{EndElement, ParseEvent, StartElement}

import scala.collection.mutable

trait Builder[T] {
  def build(): T
}

case class Item(partNumber: String)

object Item {
  def apply(builder: ItemBuilder,
            path: mutable.Stack[String]): PartialFunction[ParseEvent, Unit] = {
    case elem @ StartElement("Item", _, _, _, _) =>
      val partNumber = elem.findAttribute("PartNumber").map(_.value).getOrElse("")
      path.push(s"Item")
      builder.partNumber = partNumber
    case EndElement("Item") =>
      path.pop()
  }
}

class ItemBuilder() extends Builder[Item] {
  var partNumber = ""

  override def build(): Item =
    Item(
      partNumber = partNumber
    )

  def reset(): Unit = {
    partNumber = ""
  }
}


implicit val system = ActorSystem()
implicit val mat = ActorMaterializer()

val path: mutable.Stack[String] = new mutable.Stack[String]()

val xml =
  """<PurchaseOrder PurchaseOrderNumber="99503" OrderDate="1999-10-20">
    |<Address Type="Shipping">
    |  <Name>Ellen Adams</Name>
    |  <Street>123 Maple Street</Street>
    |  <City>Mill Valley</City>
    |  <State>CA</State>
    |  <Zip>10999</Zip>
    |  <Country>USA</Country>
    |</Address>
    |<Address Type="Billing">
    |  <Name>Tai Yee</Name>
    |  <Street>8 Oak Avenue</Street>
    |  <City>Old Town</City>
    |  <State>PA</State>
    |  <Zip>95819</Zip>
    |  <Country>USA</Country>
    |</Address>
    |<DeliveryNotes>Please leave packages in shed by driveway.</DeliveryNotes>
    |<Items>
    |  <Item PartNumber="872-AA">
    |    <ProductName>Lawnmower</ProductName>
    |    <Quantity>1</Quantity>
    |    <USPrice>148.95</USPrice>
    |    <Comment>Confirm this is electric</Comment>
    |  </Item>
    |  <Item PartNumber="926-AA">
    |    <ProductName>Baby Monitor</ProductName>
    |    <Quantity>2</Quantity>
    |    <USPrice>39.98</USPrice>
    |    <ShipDate>1999-05-21</ShipDate>
    |  </Item>
    |</Items>
    |</PurchaseOrder>""".stripMargin

val documentStream = Source.single(xml)

val builder = new ItemBuilder()

val default: PartialFunction[ParseEvent, Unit] = {
  case Characters(any) =>
  case StartElement(localName, _, _, _, _) =>
    path.push(localName)
  case EndElement(localName) =>
    path.pop()
  case any =>
}

val handle: PartialFunction[ParseEvent, Unit] = Item(builder, path) orElse
  default

val source: Source[Item, akka.NotUsed] = documentStream
  .map(ByteString(_))
  .via(XmlParsing.parser)
  .splitWhen(_ match {
    case StartElement("Item", _, _, _, _) =>
      true
    case _ =>
      false
  })
  .fold[ItemBuilder](new ItemBuilder()) {
  case (_, parseEvent) =>
    handle(parseEvent)
    builder
}
  .map { builder: ItemBuilder =>
    val item = builder.build()
    builder.reset()
    item
  }
  .concatSubstreams
  .filterNot(_.partNumber.isEmpty)

val resultFuture: Future[Seq[Item]] = source
  .runWith(Sink.seq)

val result: Seq[Item] = Await.result(resultFuture, 5.seconds)

println("items : " + result)
println("END")

该示例发布在Scastie

这种方法需要为每个标签（val 句柄：PartialFunction）提供大量处理程序，这可能容易出错且过于脆弱。

我想知道如何以更简洁的方式处理ParseEvent并将它们组合成所需的Item对象。任何建议如何避免样板代码？反序列化器有更简洁的模式吗？

score 1 · Accepted Answer

我已经稍微清理了您的代码，并扩展了构建器，以说明它如何在一系列子元素事件之后实际构建对象。也许其他人可以改进我的版本。

我建议您在解析 Item后拆分流。这样，您可以继续使用单例构建器和简单路径堆栈。

您可以为 Item 的任何假设的复杂子树添加类似的处理程序和构建器；唯一的区别是将 SubElement.build() 结果分配给某些 ItemBuilder 属性，而不是返回它。

如果您需要处理具有相同名称的子元素，您可以更深入地查看路径堆栈，或者添加额外的状态跟踪。

import akka.stream.alpakka.xml.scaladsl.XmlParsing
import akka.stream.scaladsl._

import scala.concurrent.duration._
import scala.concurrent.{Await, Future}
import akka.util.ByteString
import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.alpakka.xml.{Characters, EndElement, ParseEvent, StartElement}

import scala.collection.mutable

case class Item(partNumber: String, productName: String)

object Item {
  def apply(path: mutable.Stack[String]): PartialFunction[ParseEvent, Any] = {
    case elem @ StartElement("Item", _, _, _, _) =>
      path.push("Item")
      ItemBuilder.reset()
      ItemBuilder.partNumber = elem.findAttribute("PartNumber").map(_.value).getOrElse("")
    case Characters(text) =>
      path.top match {
        case "ProductName" => ItemBuilder.productName = text
        case _ => ()
      }
    case EndElement("Item") =>
      path.pop()
      ItemBuilder.build()
  }
}

object ItemBuilder {
  var partNumber = ""
  var productName = ""

  def build(): Item =
    Item(
      partNumber = partNumber,
      productName = productName)

  def reset(): Unit = {
    partNumber = ""
    productName = ""
  }
}

object AlpakkaDemo extends App {

  implicit val system = ActorSystem()
  implicit val mat = ActorMaterializer()

  val path: mutable.Stack[String] = new mutable.Stack[String]()

  val xml =
    """<PurchaseOrder PurchaseOrderNumber="99503" OrderDate="1999-10-20">
      |<Address Type="Shipping">
      |  <Name>Ellen Adams</Name>
      |  <Street>123 Maple Street</Street>
      |  <City>Mill Valley</City>
      |  <State>CA</State>
      |  <Zip>10999</Zip>
      |  <Country>USA</Country>
      |</Address>
      |<DeliveryNotes>Please leave packages in shed by driveway.</DeliveryNotes>
      |<Items>
      |  <Item PartNumber="872-AA">
      |    <ProductName>Lawnmower</ProductName>
      |    <Quantity>1</Quantity>
      |    <USPrice>148.95</USPrice>
      |    <Comment>Confirm this is electric</Comment>
      |  </Item>
      |  <Item PartNumber="926-AA">
      |    <ProductName>Baby Monitor</ProductName>
      |    <Quantity>2</Quantity>
      |    <USPrice>39.98</USPrice>
      |    <ShipDate>1999-05-21</ShipDate>
      |  </Item>
      |</Items>
      |</PurchaseOrder>""".stripMargin

  val defaultElementHandler: PartialFunction[ParseEvent, Any] = {
    case StartElement(localName, _, _, _, _) => path.push(localName)
    case EndElement(localName) => path.pop()
    case _ => ()
  }

  val handlersChain: PartialFunction[ParseEvent, Any] =
    Item(path) orElse
    defaultElementHandler

  val source: Source[Item, akka.NotUsed] = Source.single(xml)
    .map(ByteString(_))
    .via(XmlParsing.parser)
    .map(handlersChain)
    .collect {
      case item: Item => item
    }.splitWhen(_ => true)  // also consider mapAsyncUnordered()
    .map {
      identity  // placeholder for any subsequent heavy Item processing
    }
    .concatSubstreams

  val resultFuture: Future[Seq[Item]] = source.runWith(Sink.seq)

  val result: Seq[Item] = Await.result(resultFuture, 5.seconds)

  println("items : " + result)
  println("END")
  system.terminate()
}

这里还有一些（非常遥远的）相关文章，它们可能会激发更多关于组织基于事件的解析的想法。

http://www.ficksworkshop.com/blog/post/design-pattern-for-event-based-parsing-of-hierarchical-data

https://www.xml.com/pub/a/2003/09/17/stax.html

https://www.infoq.com/articles/HIgh-Performance-Parsers-in-Java-V2

xml - Alpakka：如何将 xml 反序列化为对象 - 反序列化器有更简洁的模式吗？

1 回答 1

Related

Reference