如何使用许多小文件加速 Spark 的 parquet 阅读器

Posted 2023-04-18

【中文标题】如何使用许多小文件加速 Spark 的 parquet 阅读器

【英文标题】：How to speed up spark's parquet reader with many small files

【发布时间】：2017-12-15 09:13:24

【问题描述】：

我正在读取一个 spark 数据帧，该数据帧以 parquet 格式存储在本地集群的 HDFS 上。 Parquet 数据分为大约 96,000 个单独的文件。现在我知道理想情况下数据不会被拆分成这么多小文件，但现在我必须以这种格式处理它。我正在使用 pyspark v2.2.0。

当我运行spark.read.parquet(data_root) 时，会发生一些奇怪的事情：spark 依次生成一系列作业，每个作业大约有 2000 个任务。它产生了 48 个这样的工作，每个工作都有一个阶段。在这 48 个作业中，它只执行了大约 96,000 个任务——我假设它为每个 parquet 文件运行一个任务。每个作业只需要大约 2 秒即可运行。

我觉得奇怪的是，这不会发生在一项有 96,000 个任务的工作中，因为那样会更快（没有阶段边界）。 2000这个数字从何而来？有没有我可以调整的参数来强制更多这些小任务进入同一个工作，从而加快速度？

【问题讨论】：

你可以先用parquet-tools把它们全部合并起来再用Spark读取。

【参考方案1】：

这是 Spark2.0 的一个新特性。 FileSourceStrategy 将较小的 parquet 文件组合成一个较大的文件，以便以优化的方式工作。理想情况下，每个 spark 执行器都希望在与 HDFS 块大小 (128MB) 相等的块大小上工作。

【讨论】：

您知道我可以使用哪些 spark 参数来调整这种情况以减少工作量吗？

我看到了一些参数，您可以调整这些参数来阻止这种情况。然而并没有使用它们。只需检查我粘贴的代码链接，您可以从 spark conf 传递一些参数

查看第 141-150 行，它看起来像 defaultMaxSplitBytes、openCostInBytes 和 bytesPerCore 通过将其设置为 min(defaultMaxSplitBytes, max(openCostInBytes, bytesPerCore)) 来控制拆分大小。如果您在答案中解释了这如何结合在一起，那么我会接受它（经过测试）。否则我会这样做并写下我自己的答案。

另外，一个理想的答案应该包括对幕后发生的事情的解释（每项工作发生了什么，以及为什么会有这么多工作）。

我尝试调整 defaultMaxSplitBytes、openCostInBytes 和 bytesPerCore（通过调整默认并行度），但这些调整都没有对问题产生任何影响。