如何使用 spark databricks xml 解析器从 Hdfs 目录加载所有 xml 文件

Posted 2023-04-17

【中文标题】如何使用 spark databricks xml 解析器从 Hdfs 目录加载所有 xml 文件

【英文标题】：How to load all xml files from a Hdfs directory using spark databricks xml parser

【发布时间】：2017-02-02 19:22:40

【问题描述】：

如何使用 databricks xml 解析器获取 Hdfs 目录中所有 xml 文件的单个数据帧，这些文件具有相同的 xml 架构

【问题讨论】：

这是一个结构不佳的问题。你应该提供一个你已经尝试过但不适合你的例子。这样你会得到更好的答案。

【参考方案1】：

您可以使用通配符来做到这一点。请参阅 Spark dataframeReader load 方法。 load 对于没有路径的数据源（即不是 HDFS 或 S3 或其他文件系统），可以采用单个路径字符串、路径序列或不带参数。 http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.DataFrameReader

val df = sqlContext.read.format("com.databricks.spark.xml")
.option("inferschema","true")
.option("rowTag", "address") //the root node of your xml to be treated as row
.load("/path/to/files/*.xml")

load 可以采用逗号分隔路径的长字符串

.load("/path/to/files/File1.xml, /path/to/files/File2.xml")

或者类似这个答案 Reading multiple files from S3 in Spark by date period

你也可以使用一系列路径

val paths: Seq[String] = ...
val df = sqlContext.read.load(paths: _*)

请注意，inferschema 对于 XML 来说非常繁忙。当涉及很多文件时，我没有取得很大的成功。指定模式效果更好。如果您可以保证您的 XML 文件都具有相同的架构，您可以使用其中的一小部分样本来推断架构，然后将其余的加载进去。我认为这并不安全，因为 XML 仍然可以是“有效的”甚至如果它缺少与 XSD 相关的某些节点或元素。

【讨论】：

如果我有多个 xml 具有不同的根标签，例如地址、部门、jobType 等，那么我该如何并行加载它们。我

@GaurangPopat 如果您有一个带有根标签address 的xml 和另一个带有根标签department 的xml，那么它们怎么可能适合相同的架构？如果您想将它们组合到一个表中，那么它们对我来说听起来像是不同的字段，或者它们可能不是根标签，您应该在 xml 路径中更高。

@Devos 让我解释一下我的情况。我有 50 个具有地址根标记的 xml 文件和其他 50 个具有部门根标记的 xml 文件。所有 100 个都存在于同一个文件夹中。我想以最有效的方式处理它们，即在加载/转换等时在集群中分配负载。部门和地址 xmls 没有关系。

@GaurangPopat 我认为你需要一个新问题来回答这个问题，但我仍然问你同样的问题，你如何期望两个不同的模式加载到同一个数据帧中？加载 2 个数据帧，然后对它们进行处理。您也可以将它们作为文本文件加载并在加载后处理架构。

【参考方案2】：

我看到您想通过分别读取每个 xml 并单独处理它们来读取 XML 数据。下面是关于它的外观的框架。

导入 scala.xml.XML

val rdd1 = sc.wholeTextFiles("/data/tmp/test/*")

val xml = rdd1.map(x=>XML.loadString(_._2.toString())

【参考方案3】：

将你的 maven 设置为 databricks 依赖项

https://mvnrepository.com/artifact/com.databricks/spark-xml_2.10/0.2.0

然后在您的 spark 程序中使用以下代码来读取 HDFS xml 文件并创建单个数据帧

导入 org.apache.spark.sql.SQLContext

val sqlContext = new SQLContext(sc)

val df = sqlContext.read .format("com.databricks.spark.xml")

.option("rowTag", "address")  //The row tag of your xml files to treat as a row

.load("file.xml")

val selectedResult = df.select("city", "zipcode")

selectedResult.write

.format("com.databricks.spark.xml")

.option("rootTag", "address") //The root tag of your xml files to treat as the root

.option("rowTag", "address")

.save("result.xml")

在 github 中找到完整的示例：

https://github.com/databricks/spark-xml/blob/master/README.md