Databricks Spark CREATE TABLE 永远需要 100 万个小 XML 文件

Question

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

实验

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

<OBSERVATION>
  <HEADER>...</HEADER>
  <RESULT>
    <ID>...</ID>
    <VALUES>...</VALUES>
  </RESULT>
</OBSERVATION>

对于解析/反序列化，我使用的是 Databricks 的 spark-xml。目前，我希望记录有两列 HEADER 和 RESULT，这就是我得到的。

CREATE TABLE Observations
USING XML
OPTIONS (
  path "/mnt/blobstorage/records/*.xml",
  rowTag "RESULT",
  rootTag "OBSERVATION",
  excludeAttribute True
)

问题

CREATE TABLE 语句运行 5.5 小时（在 Spark UI 中名为 sql at SQLDriverLocal.scala:87 的 SQL 查询），其中只有 1 小时在 Spark 中花费作业（在 Spark UI 的作业选项卡中）。

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

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

• 预定义架构（仅从第一个 XML 文件）
• 以纯文本形式摄取，无需解析（使用TEXT 提供程序）。 在这两种情况下，同样的问题仍然存在。

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

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

问题

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

Answer 1

我的猜测是您遇到了一个小文件问题，因为您只处理 15 GB。我会将每个 ca 的小文件合并到更大的文件中。 250 MB 大小。 由于您的数据集仍然很小，您可以在驱动程序上执行此操作。以下代码显示了在驱动程序节点上进行合并（不考虑最佳文件大小）：

1。将文件从 Blob 复制到本地文件系统并生成文件合并脚本：

# copy files from mounted storage to driver local storage
dbutils.fs.cp("dbfs:/mnt/blobstorage/records/", "file:/databricks/driver/temp/records", recurse=True)  

unzipdir= 'temp/records/'
gzipdir= 'temp/gzip/'

# generate shell-script and write it into the local filesystem
script = "cat " + unzipdir + "*.xml > " + gzipdir + """all.xml gzip """ + gzipdir + "all.xml"
dbutils.fs.put("file:/databricks/driver/scripts/makeone.sh", script, True)

2。运行 shell 脚本

%sh
sudo sh ./scripts/makeone.sh

3。将文件复制回挂载的存储

dbutils.fs.mv("file:/databricks/driver/" + gzipdir, "dbfs:/mnt/mnt/blobstorage/recordsopt/", recurse=True) 

另一个重要的一点是 spark-xml 库采用了两步法：

1. 它解析数据以推断架构。如果参数 samplingRatio 没有改变，它将对整个数据集执行此操作。通常只对较小的样本执行此操作就足够了，或者您可以预定义架构（为此使用参数架构），则不需要此步骤。
2. 读取数据。

最后，我建议将数据存储在 parquet 中，因此对基于列的格式进行更复杂的查询，然后直接在 xmls 上进行，并使用 spark-xml 库进行此预处理步骤。