为具有大量输入文件的 Spark SQL 作业加速 InMemoryFileIndex

Question

我有一个用 Java 编码的 apache spark sql 作业（使用数据集），它从 70,000 到 150,000 个文件中输入。

构建 InMemoryFileIndex 似乎需要 45 分钟到 1.5 小时不等。

在此期间没有日志，网络使用率非常低，并且几乎没有 CPU 使用率。

这是我在 std 输出中看到的示例：

24698 [main] INFO org.spark_project.jetty.server.handler.ContextHandler  - Started o.s.j.s.ServletContextHandler@32ec9c90{/static/sql,null,AVAILABLE,@Spark}
25467 [main] INFO org.apache.spark.sql.execution.streaming.state.StateStoreCoordinatorRef  - Registered StateStoreCoordinator endpoint
2922000 [main] INFO org.apache.spark.sql.execution.datasources.InMemoryFileIndex  - Listing leaf files and directories in parallel under: <a LOT of file url's...>
2922435 [main] INFO org.apache.spark.SparkContext  - Starting job: textFile at SomeClass.java:103

在这种情况下，有 45 分钟基本上没有发生任何事情（据我所知）。

我使用以下方法加载文件：

sparkSession.read().textFile(pathsArray)

有人可以解释 InMemoryFileIndex 中发生了什么，以及如何使这一步更快？

Answer 1

InMemoryFileIndex 负责分区发现（以及随后的分区修剪），它正在执行文件列表，并且可能会运行并行作业，如果您有很多文件，这可能需要一些时间，因为它必须为每个文件编制索引。执行此操作时，Spark 会收集有关文件的一些基本信息（例如它们的大小），以计算一些在查询计划期间使用的基本统计信息。如果您想避免每次读入数据时出现这种情况，您可以使用 Metastore 和 saveAsTable() 命令将数据保存为数据源表（Spark 2.1 支持），并且此分区发现将仅执行一次并且信息将保存在 Metastore 中。然后就可以使用metastore读取数据了

sparkSession.read.table(table_name)

并且它应该很快，因为将跳过此分区发现阶段。我建议看这个讨论这个问题的 Spark 峰会演讲。