Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

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

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?

like image 690
SimpleSam5 Avatar asked Nov 02 '18 00:11

SimpleSam5


People also ask

How can I make my Spark work faster?

Spark's efficiency is based on its ability to process several tasks in parallel at scale. Therefore, the more we facilitate the division into tasks, the faster they will be completed. This is why optimizing a Spark job often means reading and processing as much data as possible in parallel.

How do I reduce the GC time on my Spark?

To avoid full GC in G1 GC, there are two commonly-used approaches: Decrease the InitiatingHeapOccupancyPercent value ( default is 45), to let G1 GC starts initial concurrent marking at an earlier time, so that we have higher chances to avoid full GC.

Why is my Spark job so slow?

Sometimes, Spark runs slowly because there are too many concurrent tasks running. The capacity for high concurrency is a beneficial feature, as it provides Spark-native fine-grained sharing. This leads to maximum resource utilization while cutting down query latencies.

How do you increase parallelism in Spark?

One important way to increase parallelism of spark processing is to increase the number of executors on the cluster. However, knowing how the data should be distributed, so that the cluster can process data efficiently is extremely important. The secret to achieve this is partitioning in 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.

like image 133
David Vrba Avatar answered Nov 15 '22 09:11

David Vrba