[SPARK-21056][SQL] Use at most one spark job to list files in InMemoryFileIndex by bbossy · Pull Request #18269 · apache/spark

bbossy · 2017-06-11T11:56:12Z

What changes were proposed in this pull request?

This PR changes InMemoryFileIndex.listLeafFiles behaviour to launch at most one spark job to list leaf files.

##JIRA
https://issues.apache.org/jira/browse/SPARK-21056

Given partitioned file relation (e.g. parquet):
root/a=../b=../c=..
InMemoryFileIndex.listLeafFiles runs numberOfPartitions(a) times numberOfPartitions(b) spark jobs sequentially to list leaf files, if both numberOfPartitions(a) and numberOfPartitions(b) are below spark.sql.sources.parallelPartitionDiscovery.threshold and numberOfPartitions(c) is above spark.sql.sources.parallelPartitionDiscovery.threshold

Since the jobs are run sequentially, the overhead of the jobs dominates and the file listing operation can become significantly slower than listing the files from the driver.
I propose that InMemoryFileIndex.listLeafFiles should launch at most one spark job for listing leaf files.

How was this patch tested?

Adapted existing tests to match expected behaviour.

AmplabJenkins · 2017-06-11T11:57:50Z

Can one of the admins verify this patch?

bbossy · 2017-06-11T12:04:53Z

Would it be safe to use the same instance of fs for all the paths in a InMemoryFileIndex? If this is the case, I can move this back to where it was before.

Yes, it's the same instance and should be reused

Thanks! Fixed.

gatorsmile · 2017-06-12T07:05:03Z

Could you please update the PR description by copying the contents from the JIRA?

Any performance number you can share?

bbossy · 2017-06-12T12:22:40Z

@gatorsmile :

I ran a synthetic scenario to show what changes, since deploying this branch would be more involved.

I created two very simple relations on a small HDFS cluster (4 datanodes). Running spark with master local[16]. The machine has 16 physical cores plus 16 hyper threaded. The namenode is on a remote machine in the same network.

Setup:

scala> case class Foo(a: Int, b: Int, c: Int, d: Int)
defined class Foo
// manypartitions: 4 times 4 times 100 partitions. Parallel listing kicks in listing level 'c' both without and with this PR
scala> val data = for {
| a <- 1.to(4)
| b <- 1.to(4)
| c <- 1.to(100)
| } yield Foo(a,b,c,100)
data: scala.collection.immutable.IndexedSeq[Foo] = Vector(Foo(1,1,1,100), ...
scala> data.toDS.write.partitionBy("a", "b", "c").parquet("hdfs://namenode/user/bbossy/manypartitions")
// morepartitions: 10 times 10 times 100 partitions. Before this PR, 100 parallel listing jobs are spawned to list c, since each b contains more directories than the threshold. // With this PR, one parallel listing job is spawned to list all b partitions, since at level b, there are more paths to list than the threshold.
scala> val data = for {
| a <- 1.to(10)
| b <- 1.to(10)
| c <- 1.to(100)
| } yield Foo(a,b,c,1000)
data: scala.collection.immutable.IndexedSeq[Foo] = Vector(Foo(1,1,1,1000), ...
scala> data.toDS.write.partitionBy("a", "b", "c").parquet("hdfs://namenode/user/bbossy/morepartitions")

Using master branch before my commits:

scala> :pa
// Entering paste mode (ctrl-D to finish)
def time[R](block: => R): R = {
val t0 = System.currentTimeMillis()
val result = block
println("Elapsed time: " + (System.currentTimeMillis - t0) + "ms")
result
}
// Exiting paste mode, now interpreting.
time: [R](block: => R)R
scala> time(spark.read.parquet("hdfs://namenode/user/bbossy/manypartitions").collect)
Elapsed time: 6506ms
res2: Array[org.apache.spark.sql.Row] = Array([100,3,2,4], ...
scala> time(spark.read.parquet("hdfs://namenode/user/bbossy/manypartitions").collect)
Elapsed time: 2905ms
res3: Array[org.apache.spark.sql.Row] = Array([100,3,2,4], ...
scala> time(spark.read.parquet("hdfs://namenode/user/bbossy/manypartitions").collect)
Elapsed time: 2744ms
res4: Array[org.apache.spark.sql.Row] = Array([100,3,2,4], ...
scala> time(spark.read.parquet("hdfs://namenode/user/bbossy/manypartitions").collect)
Elapsed time: 2683ms
res5: Array[org.apache.spark.sql.Row] = Array([100,3,2,4], ...
scala> time(spark.read.parquet("hdfs://namenode/user/bbossy/morepartitions").collect)
Elapsed time: 16068ms
res6: Array[org.apache.spark.sql.Row] = Array([1000,6,8,31], ...
scala> time(spark.read.parquet("hdfs://namenode/user/bbossy/morepartitions").collect)
Elapsed time: 16047ms
res7: Array[org.apache.spark.sql.Row] = Array([1000,6,8,31], ...
scala> time(spark.read.parquet("hdfs://namenode/user/bbossy/morepartitions").collect)
Elapsed time: 15691ms
res8: Array[org.apache.spark.sql.Row] = Array([1000,6,8,31], ...
scala> time(spark.read.parquet("hdfs://namenode/user/bbossy/morepartitions").collect)
Elapsed time: 15767ms
res9: Array[org.apache.spark.sql.Row] = Array([1000,6,8,31], ...
scala>

UI timeline:

Using this PR:

// omitting def time...
scala> time(spark.read.parquet("hdfs://namenode/user/bbossy/manypartitions").collect)
Elapsed time: 6790ms
res0: Array[org.apache.spark.sql.Row] = Array([100,3,2,4], ...
scala> time(spark.read.parquet("hdfs://namenode/user/bbossy/manypartitions").collect)
Elapsed time: 4481ms
res1: Array[org.apache.spark.sql.Row] = Array([100,3,2,4], ...
scala> time(spark.read.parquet("hdfs://namenode/user/bbossy/manypartitions").collect)
Elapsed time: 4465ms
res2: Array[org.apache.spark.sql.Row] = Array([100,3,2,4], ...
scala> time(spark.read.parquet("hdfs://namenode/user/bbossy/manypartitions").collect)
Elapsed time: 4103ms
res3: Array[org.apache.spark.sql.Row] = Array([100,3,2,4], ...
scala> time(spark.read.parquet("hdfs://namenode/user/bbossy/morepartitions").collect)
Elapsed time: 4717ms
res4: Array[org.apache.spark.sql.Row] = Array([1000,6,8,31], ...
scala> time(spark.read.parquet("hdfs://namenode/user/bbossy/morepartitions").collect)
Elapsed time: 4434ms
res5: Array[org.apache.spark.sql.Row] = Array([1000,6,8,31], ...
scala> time(spark.read.parquet("hdfs://namenode/user/bbossy/morepartitions").collect)
Elapsed time: 5219ms
res6: Array[org.apache.spark.sql.Row] = Array([1000,6,8,31], ...
scala> time(spark.read.parquet("hdfs://namenode/user/bbossy/morepartitions").collect)
Elapsed time: 4429ms
res7: Array[org.apache.spark.sql.Row] = Array([1000,6,8,31], ...
scala>

UI timeline:

Is there something more specific that I should look into?

bbossy · 2017-06-13T08:25:43Z

ping @gatorsmile @srowen and possibly @cloud-fan : Would like to hear your thoughts on this.

cloud-fan · 2017-06-14T02:46:29Z

how about
if (paths.isEmpty) { Nil } else { val fs = paths.head.getFileSystem(hadoopConf) ...... }

cloud-fan · 2017-06-14T02:48:09Z

can we merge these flatMaps?

something like
paths.flatMap { path => try { val status = fs.get.listStatus(path) val filteredStatuses = statuses.filterNot(status => shouldFilterOut(status.getPath.getName)) .... } catch ... }

then we can still keep the previous code structure

kiszk · 2017-06-15T13:15:34Z

nit: style issue. this if-then-else should be moved to left with 2 spaces

mallman · 2017-06-16T17:00:44Z

@bbossy I've built and deployed a branch of Spark 2.2 with your patch and compared its behavior to the same branch of Spark 2.2 without your patch. I'm seeing different behavior, but not what I expected.

My test table has three partition columns, ds, h and chunk. There are 1 ds values, 2 h values, and 51 chunk values, split into 27 and 24 partitions under the two h directories. I set spark.sql.sources.parallelPartitionDiscovery.threshold to 10. I believe this fits the scenario you're trying to remedy.

I use spark.read.parquet to load the table. When I load the table using the unpatched branch, Spark launches three jobs with 27, 24 and 1 stages, in that order. When I load the table using the patched branch, Spark launches three jobs with 51, 1 and 51 stages, in that order. Does this match your expectations? I was expecting to see Spark launch two jobs.

…ndex

bbossy · 2017-06-18T14:07:43Z

@mallman I'm not sure where this difference in behaviour is coming from. The following test in FileIndexSuite passes:

 test("mallman's scenario") {
withSQLConf(SQLConf.PARALLEL_PARTITION_DISCOVERY_THRESHOLD.key -> "10") {
withTempDir { dir =>
for (chunk <- 1.to(27)) {
val f = new File(dir, s"ds=1/h=one/chunk=$chunk/file.parquet")
f.getParentFile.mkdirs()
f.createNewFile()
}
for (chunk <- 1.to(24)) {
val f = new File(dir, s"ds=1/h=two/chunk=$chunk/file.parquet")
f.getParentFile.mkdirs()
f.createNewFile()
}
HiveCatalogMetrics.reset()
assert(HiveCatalogMetrics.METRIC_PARALLEL_LISTING_JOB_COUNT.getCount() == 0)
new InMemoryFileIndex(spark, Seq(new Path(dir.getCanonicalPath)), Map.empty, None)
assert(HiveCatalogMetrics.METRIC_PARALLEL_LISTING_JOB_COUNT.getCount() == 1)
}
}
}

Does it match your scenario?

I'll dig around a bit later to see if I can come up with an explanation.

bbossy · 2017-06-18T14:08:24Z

@cloud-fan Could you take another look, please?

cloud-fan · 2017-06-19T05:10:38Z

+ try {
+ val fStatuses = fs.listStatus(path)
+ val filtered = fStatuses.filterNot(status => shouldFilterOut(status.getPath.getName))
+ if (filtered.nonEmpty) {


nit: filtered.map(path -> _), so that we don't need the if-else here, and the flatMapthere

cloud-fan · 2017-06-19T05:13:37Z

+ fStatuses.map { f => path -> f }
+ }.partition { case (_, fStatus) => fStatus.isDirectory }
+ val pathsToList = dirs.map { case (_, fStatus) => fStatus.getPath }
+ val nestedFiles = if (pathsToList.nonEmpty) {


do we need this if check?

cloud-fan · 2017-06-19T05:19:58Z

let's wait @mallman 's response to make sure this patch does fix the problem

mallman · 2017-06-21T17:21:42Z

Hi @bbossy

Does it match your scenario?

It does not match my scenario. I'm reading files from HDFS. In your test, you're reading files from the local filesystem. Can you try a test using files stored in HDFS?

Also, I'm not testing with InMemoryFileIndex directly. Rather, I'm using spark.read.parquet. Can you try a test with that scenario?

HyukjinKwon · 2017-07-24T03:06:59Z

gentle ping @bbossy, I just want to be sure if it is in progress in any way.

bbossy · 2017-07-24T15:17:40Z

@HyukjinKwon I'll see that I can address the outstanding review comments in the next day or two.

## What changes were proposed in this pull request? This PR proposes to close stale PRs, mostly the same instances with apache#18017Closesapache#14085 - [SPARK-16408][SQL] SparkSQL Added file get Exception: is a directory … Closesapache#14239 - [SPARK-16593] [CORE] [WIP] Provide a pre-fetch mechanism to accelerate shuffle stage. Closesapache#14567 - [SPARK-16992][PYSPARK] Python Pep8 formatting and import reorganisation Closesapache#14579 - [SPARK-16921][PYSPARK] RDD/DataFrame persist()/cache() should return Python context managers Closesapache#14601 - [SPARK-13979][Core] Killed executor is re spawned without AWS key… Closesapache#14830 - [SPARK-16992][PYSPARK][DOCS] import sort and autopep8 on Pyspark examples Closesapache#14963 - [SPARK-16992][PYSPARK] Virtualenv for Pylint and pep8 in lint-python Closesapache#15227 - [SPARK-17655][SQL]Remove unused variables declarations and definations in a WholeStageCodeGened stage Closesapache#15240 - [SPARK-17556] [CORE] [SQL] Executor side broadcast for broadcast joins Closesapache#15405 - [SPARK-15917][CORE] Added support for number of executors in Standalone [WIP] Closesapache#16099 - [SPARK-18665][SQL] set statement state to "ERROR" after user cancel job Closesapache#16445 - [SPARK-19043][SQL]Make SparkSQLSessionManager more configurable Closesapache#16618 - [SPARK-14409][ML][WIP] Add RankingEvaluator Closesapache#16766 - [SPARK-19426][SQL] Custom coalesce for Dataset Closesapache#16832 - [SPARK-19490][SQL] ignore case sensitivity when filtering hive partition columns Closesapache#17052 - [SPARK-19690][SS] Join a streaming DataFrame with a batch DataFrame which has an aggregation may not work Closesapache#17267 - [SPARK-19926][PYSPARK] Make pyspark exception more user-friendly Closesapache#17371 - [SPARK-19903][PYSPARK][SS] window operator miss the `watermark` metadata of time column Closesapache#17401 - [SPARK-18364][YARN] Expose metrics for YarnShuffleService Closesapache#17519 - [SPARK-15352][Doc] follow-up: add configuration docs for topology-aware block replication Closesapache#17530 - [SPARK-5158] Access kerberized HDFS from Spark standalone Closesapache#17854 - [SPARK-20564][Deploy] Reduce massive executor failures when executor count is large (>2000) Closesapache#17979 - [SPARK-19320][MESOS][WIP]allow specifying a hard limit on number of gpus required in each spark executor when running on mesos Closesapache#18127 - [SPARK-6628][SQL][Branch-2.1] Fix ClassCastException when executing sql statement 'insert into' on hbase table Closesapache#18236 - [SPARK-21015] Check field name is not null and empty in GenericRowWit… Closesapache#18269 - [SPARK-21056][SQL] Use at most one spark job to list files in InMemoryFileIndex Closesapache#18328 - [SPARK-21121][SQL] Support changing storage level via the spark.sql.inMemoryColumnarStorage.level variable Closesapache#18354 - [SPARK-18016][SQL][CATALYST][BRANCH-2.1] Code Generation: Constant Pool Limit - Class Splitting Closesapache#18383 - [SPARK-21167][SS] Set kafka clientId while fetch messages Closesapache#18414 - [SPARK-21169] [core] Make sure to update application status to RUNNING if executors are accepted and RUNNING after recovery Closesapache#18432 - resolve com.esotericsoftware.kryo.KryoException Closesapache#18490 - [SPARK-21269][Core][WIP] Fix FetchFailedException when enable maxReqSizeShuffleToMem and KryoSerializer Closesapache#18585 - SPARK-21359 Closesapache#18609 - Spark SQL merge small files to big files Update InsertIntoHiveTable.scala Added: Closesapache#18308 - [SPARK-21099][Spark Core] INFO Log Message Using Incorrect Executor I… Closesapache#18599 - [SPARK-21372] spark writes one log file even I set the number of spark_rotate_log to 0 Closesapache#18619 - [SPARK-21397][BUILD]Maven shade plugin adding dependency-reduced-pom.xml to … Closesapache#18667 - Fix the simpleString used in error messages Closesapache#18782 - Branch 2.1 Added: Closesapache#17694 - [SPARK-12717][PYSPARK] Resolving race condition with pyspark broadcasts when using multiple threads Added: Closesapache#16456 - [SPARK-18994] clean up the local directories for application in future by annother thread Closesapache#18683 - [SPARK-21474][CORE] Make number of parallel fetches from a reducer configurable Closesapache#18690 - [SPARK-21334][CORE] Add metrics reporting service to External Shuffle Server Added: Closesapache#18827 - Merge pull request 1 from apache/master ## How was this patch tested? N/A Author: hyukjinkwon <gurwls223@gmail.com> Closesapache#18780 from HyukjinKwon/close-prs.

bbossy commented Jun 11, 2017
View reviewed changes

cloud-fan reviewed Jun 14, 2017
View reviewed changes

kiszk reviewed Jun 15, 2017
View reviewed changes

bbossy added 3 commits June 18, 2017 13:56

SPARK-21056: use at most one spark job to list files in InMemoryFileI…
68155cc
…ndex

SPARK-21056: Use the same instance of FileSystem
2150f1c

SPARK-21056: Reorg code
579c567

bbossy force-pushed the SPARK-21056 branch from 97b2251 to 579c567Compare June 18, 2017 12:00

bbossy added 2 commits June 18, 2017 14:03

SPARK-21056: Bail out early if paths is empty
cb21cf7

SPARK-21056: Fix serialization issue in bulkListLeafFiles
c4fba40

cloud-fan reviewed Jun 19, 2017
View reviewed changes

HyukjinKwon mentioned this pull request Jul 31, 2017
[INFRA] Close stale PRs #18780
Closed

asfgit closed this in 3a45c7fAug 5, 2017

Uh oh!

Conversation

bbossy commented Jun 11, 2017• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

AmplabJenkins commented Jun 11, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gatorsmile commented Jun 12, 2017

Uh oh!

bbossy commented Jun 12, 2017

Setup:

Using master branch before my commits:

Using this PR:

Uh oh!

bbossy commented Jun 13, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kiszkJun 15, 2017• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mallman commented Jun 16, 2017

Uh oh!

bbossy commented Jun 18, 2017

Uh oh!

bbossy commented Jun 18, 2017

Uh oh!

cloud-fanJun 19, 2017• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan commented Jun 19, 2017

Uh oh!

mallman commented Jun 21, 2017

Uh oh!

HyukjinKwon commented Jul 24, 2017

Uh oh!

bbossy commented Jul 24, 2017

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

8 participants

bbossy commented Jun 11, 2017•
edited
Loading

kiszkJun 15, 2017•
edited
Loading

cloud-fanJun 19, 2017•
edited
Loading