问题标签 [apache-spark-xml]

来自火花文件

我定义了读取 xml 记录的函数

如果我想将 xml 格式的记录传递给.load函数，而不是从文件中加载，我该怎么办，比如.load(inputStreamXML)

我还有一个关于 Spark 和 Scala 的问题。我想使用该技术来获取数据并生成 xml。因此，我想知道是否可以自己创建节点（不是自动创建）以及我们可以使用什么库？我搜索但没有发现什么很有趣的东西（就像我是这项技术的新手，我不知道很多关键字）。我想知道 Spark 中是否有类似这样的代码（我在 scala 中编写。它在本地工作，但我不能在 Spark 中使用 new File()）。

我想知道火花是否可以做到这一点。

感谢您的回答，祝您有美好的一天。

我有一组100 万个XML 文件，每个文件在 Azure Blob 存储中的大小约为 14KB，安装在 Azure Databricks 中，我正在尝试使用CREATE TABLE，期望每个文件都有一条记录。

本实验

文件的内容结构如下所示。为了简单和性能实验，除了<ID>元素之外的所有文件内容都保持相同。

对于解析/反序列化，我使用Databricks的 spark-xml。此刻，我期待有两列HEADER和的记录RESULT，这就是我得到的。

问题

该CREATE TABLE语句运行5.5 小时（在 Spark UI 中具有名称的 SQL 查询sql at SQLDriverLocal.scala:87），其中只有1 小时用于 Spark 作业（在 Spark UI 的“作业”选项卡中）。

我注意到带有CREATE TABLE命令的单元格Listing files at "/mnt/blobstorage/records/*.xml"大部分时间都卡在上面。首先，我认为这是存储连接器中的扩展问题。但是，我可以在~25 秒内对~500K大小相似的 JSON 文件运行命令（XML 与 JSON 的问题？）。

我也知道spark-xml读取所有文件以推断架构，这可能是瓶颈。为了消除这种可能性，我尝试：

• 预定义模式（仅来自第一个 XML 文件）
• 以纯文本形式摄取而不进行解析（使用TEXT提供程序）。在这两种情况下，同样的问题仍然存在。

对于10K记录，相同的语句在20秒内运行，对于200K记录在30 分钟内运行。使用线性缩放（这显然不会发生），100 万条记录将在~33 分钟内完成。

我的 Databricks 集群有 1 个工作节点和 3 个驱动程序节点，每个节点都有256 GB的 RAM 和64 个内核，因此不应该存在缓存瓶颈。我已经在 4 天内多次运行成功地重现了该问题。

问题

我在这里做错了什么？如果在 期间我可以做一些分区/集群CREATE TABLE，我该怎么做？

使用 spark，我试图从路径中读取一堆 xml，其中一个文件是一个不是 xml 的虚拟文件。

我希望火花告诉我一个特定的文件是无效的，无论如何

添加“badRecordsPath”otiton 将坏数据写入 JSON 文件的指定位置，但同样不适用于 xml，还有其他方法吗？

我正在尝试将 df 编写为 xml 文件。我的问题是如何在 xml 输出中执行分组。

我的输入数据为 csv 文件：

我正在读取 csv 文件并作为 xml 文件写回，我想知道在编写 xml 时如何指定嵌套模式。

目前我能得到什么

预期 o/p

我正在尝试使用 Databricks spark-xml 包将 XML 文件读入 Spark 数据帧。但是，当遇到处理指令时，Spark 会引发一个错误，声称发生了意外事件。

我正在尝试将 XML 文件导入数据帧，然后我可以将其操作为平面文件以写入 CSV。数据集足够大，我们需要某种处理程序，例如 Spark。我查看了 spark-xml 文档，找不到任何关于处理指令的提及。我实际上不需要说明中的任何信息，所以如果可以选择的话，我很乐意忽略它们，但因为它们正在干扰整个文件。任何建议，将不胜感激。

下面是重现问题的 XML 片段：

这是我尝试在 Python 中读取 XML 的方式：

为了完整起见，以下是我加载数据块并将脚本提交给 Spark 的方式：

当我尝试使用上面的代码读取 XML 时，Spark 会引发一个异常，声称是“意外事件”。在下面找到确切的错误消息。

我正在使用 Spark XML 解析包含一些用户定义实体的大型文档。这是文件中的一个简单片段

可以在 XML 文档中找到的内联 DTD 中正确定义了实体，例如这里

但是，在模式检测阶段解析失败......

原因似乎是用户定义的实体：当我转义它们（例如&unc;）时，一切都会再次运行。

我该如何解决这个问题？

目前，我正在使用 azure-storage-blob 和 hadoop-azure 包将文件从 Blob 存储下载到本地。

这些下载的文件实际上是 XML 文件。然后，我必须处理每个内容。为此，我使用 spark-xml_2.11 (com.databricks.spark.xml) 包。

load() 方法需要一个路径（由本地或分布式文件系统支持的数据）。那么，是否可以直接从 Blob 存储中加载它们？

我找到了本指南https://docs.databricks.com/spark/latest/data-sources/azure/azure-storage.html但是第一个选项，将 Azure Blob 存储容器装载到 DBFS，需要一个 Databrick 集群。

对于第二个选项，直接访问 Azure Blob 存储，我之前测试过设置帐户访问密钥。

但是引发了以下异常：

此外，我尝试将协议更改为 wasbs，但再次引发了类似的异常：

请问，有什么建议或意见吗？

假设我有以下 xml 数据：

它的架构是：

作为根标签"students"。

在这里，我想更新某些列的值。

我想使用更新"studentId"列的值UDF。我找到了一种方法：

然后，我想更新一个嵌套列，即"info.rollNo". 应用上述过程给了我另一个新列“ <info.rollNo>updated_value</info.rollNo>”。找了一阵子，找到了一个办法：

对于第三个嵌套列，我尝试了上述方式。但我无法弄清楚这个过程。在这里，问题是，有人可以解释一下更新嵌套列值的算法，其嵌套级别可能是 3、4、5 等等。例如：我想更新以下字段。 "info.address.permanent"哪个是结构， "subjects.subject.credit"哪个是数组的元素"subject"

PS：如果您知道任何其他更新某些列的方法，请提及。

场景：我通过指定加载模式读取了两个 XML 文件。

在模式中，其中一个标签是强制性的。一个 XML 缺少该强制标记。

现在，当我执行以下操作时，我希望过滤掉缺少强制标记的 XML。

在代码中，当我尝试计算数据集的行数时，我得到的计数为 2（2 个输入 XMLS），但是当我尝试通过 show() 方法打印数据集时，我得到一个 NPE。

当我调试上述行并执行以下操作时，我得到 0 作为计数。

问题：

谁能回答下面的问题/确认我的理解

1. 为什么 spark Dataset 不过滤没有强制列的行？
2. 为什么 count 中没有异常，但 show 方法中没有异常？

对于 2，我相信计数只是在不查看内容的情况下计算行数。为了展示，迭代器实际上通过 Struct Fields 打印它们的值，当它没有找到强制列时，它会出错。

PS 如果我将必填列设为可选，则一切正常。

编辑：

根据要求提供阅读选项

为了加载数据，我正在执行以下操作

按要求提供样品

架构：

示例 XML：

JSON 格式的架构：

我的场景可以通过将元素“旧”设置为 nullable = false 并从 XML 中删除标签来重现