Speed up InMemoryFileIndex for Spark SQL job with large number of input files

user_1052158 0
Viewed 4807

I have an apache spark sql job (using Datasets), coded in Java, that get's it's input from between 70,000 to 150,000 files.

It appears to take anywhere from 45 minutes to 1.5 hours to build the InMemoryFileIndex.

There are no logs, very low network usage, and almost no CPU usage during this time.

Here's a sample of what I see in the std output:

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

In this case, there was 45 minutes of essentially nothing happening (as far as I could tell).

I load the files using:

sparkSession.read().textFile(pathsArray)

Can someone explain what is going on in InMemoryFileIndex, and how can I make this step faster?

apache-spark-sql apache-spark
1 Answers

The InMemoryFileIndex is responsible for partition discovery (and consequently partition pruning), it is doing file listing and it may run a parallel job which can take some time if you have a lot of files, since it has to index each file. When doing this, Spark collects some basic information about the files (their size for instance) to compute some basic statistics that are than used during query planning. If you want to avoid this each time you read the data in, you can save the data as a datasource table (it is supported from Spark 2.1) using the metastore and the saveAsTable() command and this partition discovery will be performed only once and the information will be kept in the metastore. Then you can read the data using the metastore

sparkSession.read.table(table_name)

and it should be fast since this partition discovery phase will be skipped. I recommend to see this Spark Summit talk in which this problem is discussed.

user_102985510
Related