Menu
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?
690
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.
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.
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.
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.
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.
133
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With
