xml - 使用 Spark/Scala 从 XML 记录中提取元素

Question

我正在尝试从 XML 记录中提取元素，其中每个 xml 文件都有许多 XML 记录。下面是我正在使用的修改后的代码和示例 xml。

我期待一个字符串数组，其中数组的每个元素都是 "user:id" 但结果是 ":"。我期望 XML.loadString 解析每个文件，结果将是单独的 XML 记录。这意味着如果我以两个示例文件为例，我最终会得到 4 个 XML 记录。事实上，它是两个。

在获取 next 后添加 println(d) 后，我得到的是表示文件的整个字符串，这可能就是 getId 的原因> 和 getUser 函数没有返回任何内容。

我是否错误地处理了负载？

import org.apache.spark.{SparkConf, SparkContext}
import scala.xml._
import scala.collection.mutable.ArrayBuffer

object Details {

    def getDetails(xmlstring: String): Iterator[Node] = {
        val nodes = XML.loadString(xmlstring)
        nodes.toIterator
    }

    def getId(detail: Node): String = {
        (detail \ "id").text
    }

    def getUser(detail: Node): String = {
        (detail \ "user").text
    }

    def getDetailList(details: Iterator[Node]): Array[String] = {
        var list = ArrayBuffer[String]()
        while (details.hasNext) {
            val d = details.next
            val user = getUser(d)
            val id = getId(d)
            val formattedText = user + ":" + id
            list += formattedText
        }
        list.toArray
    }

    def main(args: Array[String]) {

        val conf = new SparkConf().setAppName("Details")
        val sc: SparkContext = new SparkContext(conf)

        val lines = sc.wholeTextFiles("file:///path/to/files/")
        val xmlStrings = lines.map(line => line._2)
        val detailsRecords = xmlStrings.map(getDetails)
        val detailsList = detailsRecords.map(getDetailList)

        spark.stop()
    }
}

还有两个示例文件...

test.xml

<details>
  <detail>
    <user>Dan</user>
    <id>5555</id>
  </detail>
  <detail>
    <user>Mike</user>
    <id>6666</id>
  </detail>
</details>

test2.xml

<details>
  <detail>
    <user>John</user>
    <id>1234</id>
  </detail>
  <detail>
    <user>Joe</user>
    <id>5678</id>
  </detail>
</details>

Answer 1

您应该使用XML for Spark .

使用此库，您可以像这样读取所有 xml 文件:

import org.apache.spark.sql.SQLContext

val sqlContext = new SQLContext(sc)

val df = sqlContext.read
   .format("com.databricks.spark.xml")
   .option("rowTag", "detail")
   .load("/home/path-with-xml-files")

这会生成一个具有架构的 DataFrame:

+----+----+
|  id|user|
+----+----+
|5555| Dan|
|6666|Mike|
|1234|John|
|5678| Joe|
+----+----+

然后从这个 DF 获取一个数组:

val id_users_array = df.collect

该数组的类型为:

id_users_array: Array[org.apache.spark.sql.Row] = Array([5555,Dan], [6666,Mike], [1234,John], [5678,Joe])

如果您只想打印 ids:

id_users_array.map(r => r.get(0)).foreach(println)

输出:

5555
6666
1234
5678

希望这有帮助。