From 6b68d61cf31748a088778dfdd66491b2f89a3c7b Mon Sep 17 00:00:00 2001
From: Liang-Chi Hsieh <viirya@gmail.com>
Date: Thu, 25 May 2017 09:55:45 +0800
Subject: [PATCH] [SPARK-20848][SQL][FOLLOW-UP] Shutdown the pool after reading
parquet files
## What changes were proposed in this pull request?
This is a follow-up to #18073. Taking a safer approach to shutdown the pool to prevent possible issue. Also using `ThreadUtils.newForkJoinPool` instead to set a better thread name.
## How was this patch tested?
Manually test.
Please review http://spark.apache.org/contributing.html before opening a pull request.
Author: Liang-Chi Hsieh <viirya@gmail.com>
Closes #18100 from viirya/SPARK-20848-followup.
---
.../parquet/ParquetFileFormat.scala | 42 ++++++++++---------
1 file changed, 22 insertions(+), 20 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
index 29ed890613..87fbf8b1bc 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
@@ -50,7 +50,7 @@ import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources._
import org.apache.spark.sql.types._
-import org.apache.spark.util.SerializableConfiguration
+import org.apache.spark.util.{SerializableConfiguration, ThreadUtils}
class ParquetFileFormat
extends FileFormat
@@ -479,27 +479,29 @@ object ParquetFileFormat extends Logging {
partFiles: Seq[FileStatus],
ignoreCorruptFiles: Boolean): Seq[Footer] = {
val parFiles = partFiles.par
- val pool = new ForkJoinPool(8)
+ val pool = ThreadUtils.newForkJoinPool("readingParquetFooters", 8)
parFiles.tasksupport = new ForkJoinTaskSupport(pool)
- parFiles.flatMap { currentFile =>
- try {
- // Skips row group information since we only need the schema.
- // ParquetFileReader.readFooter throws RuntimeException, instead of IOException,
- // when it can't read the footer.
- Some(new Footer(currentFile.getPath(),
- ParquetFileReader.readFooter(
- conf, currentFile, SKIP_ROW_GROUPS)))
- } catch { case e: RuntimeException =>
- if (ignoreCorruptFiles) {
- logWarning(s"Skipped the footer in the corrupted file: $currentFile", e)
- None
- } else {
- throw new IOException(s"Could not read footer for file: $currentFile", e)
+ try {
+ parFiles.flatMap { currentFile =>
+ try {
+ // Skips row group information since we only need the schema.
+ // ParquetFileReader.readFooter throws RuntimeException, instead of IOException,
+ // when it can't read the footer.
+ Some(new Footer(currentFile.getPath(),
+ ParquetFileReader.readFooter(
+ conf, currentFile, SKIP_ROW_GROUPS)))
+ } catch { case e: RuntimeException =>
+ if (ignoreCorruptFiles) {
+ logWarning(s"Skipped the footer in the corrupted file: $currentFile", e)
+ None
+ } else {
+ throw new IOException(s"Could not read footer for file: $currentFile", e)
+ }
}
- } finally {
- pool.shutdown()
- }
- }.seq
+ }.seq
+ } finally {
+ pool.shutdown()
+ }
}
/**
--
GitLab