Alpakka:如何对XML进行应对对象的序列化 - 对避难所的质量是否有更简洁的模式

Question

我具有复杂的XML数据(它可能包含很多数据，并且可能具有15GB以上(，该数据具有具有深层结构的复杂性质。我们需要为巨大的XML进行流处理。使用新的Alpakka库是我们的首选，因为它是一个有前途的解决方案。

Scala-XML序列化和其他Scala库上有过时的线程，但是我们需要处理大量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 对象中。有什么建议如何避免样板代码？是否有更简洁的模式？

Answer 1

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

我建议您在之后将流拆分。这样，您可以继续使用Singleton Builder和简单的路径堆栈。

您可以为项目的任何假设的复杂子树添加类似的处理程序和建筑商；唯一的区别是将sublement.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-----------------------------------

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

https://www.infoq.com/articles/high-performance-parsers-in-java-v2