使用多个 RowTag 在 Spark 中读取 XML 文件答案

【问题标题】：Read XML File in Spark with multiple RowTags使用多个 RowTag 在 Spark 中读取 XML 文件
【发布时间】：2019-04-02 08:40:57
【问题描述】：

我想将一个包含 3 个不同 RowTag 的巨大 XML 文件读入 Apache Spark 数据帧。

RowTag = XML 元素，您在 Spark 中将其解释为一行。

标签

包含不同的数据结构
不重叠

xml-spark (https://github.com/databricks/spark-xml) 只提供一次读取一个 RowTag，所以我需要读取相同的文件 3 次（效率不高）。

有什么方法可以一次读取文件吗？

详情：

我有一个巨大的 XML 文件 (24 GB)，其中包含 3 个列表：

<myFile>
    <ContainedResourceList>
        <SoundRecording><Title>A</Title></SoundRecording>
      ... several million records ...
        <SoundRecording><Title>Z</Title></SoundRecording>
    </ContainedResourceList>

    <ContainedReleaseList>
        <Release><ReleaseType>Single</ReleaseType></Release>
      ... several million records ...
        <Release><ReleaseType>LP</ReleaseType></Release>
    </ContainedReleaseList>

    <ContainedTransactionList>
        <Transaction><Sales>1</Sales></Transaction>
      ... several million records ...
        <Transaction><Sales>999</Sales></Transaction>
    </ContainedTransactionList>
</myFile>

XML 文件有效。我想阅读 RowTags SoundRecording、Release & Transaction。

我更喜欢 Scala 库，但我会为任何支持读取的库感到高兴。

PS： 输出和他的架构是什么样的？

最佳选择：一个包含 3 个 DataFrame 的数组，每个 RowTag 一个
丑陋的选项：一个 DataFrame 包含所有 3 个数据结构的可能元素

【问题讨论】：

标签： spark-dataframe databricks apache-spark-xml

【解决方案1】：

一种简单的方法是使用explode 函数。您可以读取将 rowTag 设置为 ContainedResourceList 的完整 xml，然后使用生成的数据框用新列分解数据框

df.withColumn("soundRec", explode($"SoundRecording"))

您可以为每个要爆炸的标签添加多个列

【讨论】：

感谢您提出的解决方案。建议的“爆炸”解决方案适用于小文件。但它在一个巨大的文件（24 GB）上失败：java.lang.OutOfMemoryError: Requested array size exceeds VM Limit 在这个解决方案中，整个 ContainedResourceList 被读入一个记录。 ContainedResourceList 包含 ca。 1000 万条录音。 ContainedResourceList 大小约为。输入文件的 1/3。我在每个 8GB RAM 的 3 个主机集群上进行了测试。我可以使用更多的 RAM，但是使用大量 RAM 来在一台机器上读取整个文件并不是使用 Spark 背后的想法。

【解决方案2】：

根据我对 spark-xml 的使用，我知道它需要 XML 文件中的 2 个标签，

根标签
行标签

您的输入文件应如下所示，

<root>
    <row>
        <FirstField> abc </FirstField>
        <SecondField> def <SecondField>
    </row>
    <row>
        <FirstField> ghi </FirstField>
        <SecondField> jkl <SecondField>
    </row>
    .
    .
    <row>
        <FirstField> uvw </FirstField>
        <SecondField> xyz <SecondField>
    </row>
</root>

要阅读上面的文件，语法是

spark-shell --packages com.databricks:spark-xml_2.11:0.5.0
import com.databricks.spark.xml._
import org.apache.spark.sql.types._
val schema = StructType(List(StructField("FirstField",StringType,true),StructField("SecondField",StringType,true)))
val df = spark.read.option("rootTag","root").option("rowTag","row").schema(schema)xml("pathToFile")

在您的情况下，您有一个 rootTag 作为“myFile”，但现在有行标签。因此，您可以尝试使用“myFile”作为 rowTag，关键是您必须按如下方式创建架构，

val schema = StructType(List(StructField("ContainedResourceList",StringType,true),StructField("ContainedReleaseList",StringType,true),StructField("ContainedTransactionList",StringType,true)))

然后读取文件，

 val df = spark.read.option("myFile","row").schema(schema).xml("pathToFile")

现在你可以处理这个df了

您有 SoundRecording、Release 和 Transaction 的重复标签，如果您为这些定义模式，那么只有重复中的第一个值被解析，在您的情况下是 <Title>A</Title>, <ReleaseType>Single</ReleaseType>, <Sales>1</Sales>。

我还没有弄清楚如何解析 Spark-xml 中的重复标签

【讨论】：

【解决方案3】：

将 myfile 读取为行标记将产生一个 HUGE 行，然后分解成行将由一个 spark worker 完成。

您可以将它们读入 3 个不同的数据帧，指定不同的行标签，因为每个数据帧都有不同的架构。这将产生 3 个不同的数据帧，每个数据帧都有数百万行，这将通过 spark 更有效地工作。

为了加快处理速度，您可以将 xml 文件预先分割成多个块，甚至进一步分割成 3 组文件 (Splitting XML file into multiple at given tags)。这样，工人可以并行读取多个部分，当他们完成一个部分时，他们可以移动到下一个部分。否则，只有一名工作人员必须按顺序读取文件并使用其自己的分区方式并将它们分发给工作人员。

然后您可以使用 spark-sql 将它们连接在一起，这就是您想要做的。

【讨论】：