Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[SPARK-20848][SQL] Shutdown the pool after reading parquet files #18073

Closed
wants to merge 3 commits into from
Closed

[SPARK-20848][SQL] Shutdown the pool after reading parquet files #18073

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e4940b9
Shutdown the pool after reading parquet files.
viirya May 23, 2017
14e09aa
Address comment.
viirya May 24, 2017
7e57595
Remove hacky test.
viirya May 24, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions ...src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -479,8 +479,9 @@ object ParquetFileFormat extends Logging {
partFiles: Seq[FileStatus],
ignoreCorruptFiles: Boolean): Seq[Footer] = {
val parFiles = partFiles.par
parFiles.tasksupport = new ForkJoinTaskSupport(new ForkJoinPool(8))
parFiles.flatMap { currentFile =>
val readParquetTaskSupport = new ForkJoinTaskSupport(new ForkJoinPool(8))
parFiles.tasksupport = readParquetTaskSupport
val footers = parFiles.flatMap { currentFile =>
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You could probably put the shutdown in a finally block to avoid adding this new reference. I think you can rearrange to use the existing one below. It does help shut it down in case of an error too. Also I think you can hold a reference just to the ForkJoinPool in order to shut it down rather than a ref to ForkJoinTaskSupport. No big deal

try {
// Skips row group information since we only need the schema.
// ParquetFileReader.readFooter throws RuntimeException, instead of IOException,
Expand All @@ -497,6 +498,8 @@ object ParquetFileFormat extends Logging {
}
}
}.seq
readParquetTaskSupport.forkJoinPool.shutdown()
footers
}

/**
Expand Down
16 changes: 16 additions & 0 deletions ...est/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormatSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,22 @@ import org.apache.spark.sql.test.SharedSQLContext

class ParquetFileFormatSuite extends QueryTest with ParquetTest with SharedSQLContext {

test("Number of threads doesn't grow extremely after parquet file reading") {
withTempDir { dir =>
val file = dir.toString + "/file"
spark.range(1).toDF("a").coalesce(1).write.parquet(file)
spark.read.parquet(file)
val numThreadBefore = Thread.activeCount
(1 to 100).map { _ =>
spark.read.parquet(file)
}
val numThreadAfter = Thread.activeCount
// Hard to test a correct thread number,
// but it shouldn't increase more than a reasonable number.
assert(numThreadAfter - numThreadBefore < 20)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

after waiting for enough time, can we expect this to be 0?

Copy link
Member Author

@viirya viirya May 24, 2017 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It reduces to a few (about 3) after waiting an enough time. The number returned by Thread.activeCount is only an estimate. So we may not expect this to be 0.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this looks hacky, can we think of a better way to test it? If not, I suggest to remove this test, as the fix is straightforward and we can verify it manually by some profile tools.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK. Let's remove the test.

}
}

test("read parquet footers in parallel") {
def testReadFooters(ignoreCorruptFiles: Boolean): Unit = {
withTempDir { dir =>
Expand Down