Skip to content

Commit f60b0b4

Browse files
committed
Safer approach to shutdown the pool.
1 parent a647466 commit f60b0b4

1 file changed

Lines changed: 22 additions & 20 deletions

File tree

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala

Lines changed: 22 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@ import org.apache.spark.sql.execution.datasources._
5050
import org.apache.spark.sql.internal.SQLConf
5151
import org.apache.spark.sql.sources._
5252
import org.apache.spark.sql.types._
53-
import org.apache.spark.util.SerializableConfiguration
53+
import org.apache.spark.util.{SerializableConfiguration, ThreadUtils}
5454

5555
class ParquetFileFormat
5656
extends FileFormat
@@ -479,27 +479,29 @@ object ParquetFileFormat extends Logging {
479479
partFiles: Seq[FileStatus],
480480
ignoreCorruptFiles: Boolean): Seq[Footer] = {
481481
val parFiles = partFiles.par
482-
val pool = new ForkJoinPool(8)
482+
val pool = ThreadUtils.newForkJoinPool("readingParquetFooters", 8)
483483
parFiles.tasksupport = new ForkJoinTaskSupport(pool)
484-
parFiles.flatMap { currentFile =>
485-
try {
486-
// Skips row group information since we only need the schema.
487-
// ParquetFileReader.readFooter throws RuntimeException, instead of IOException,
488-
// when it can't read the footer.
489-
Some(new Footer(currentFile.getPath(),
490-
ParquetFileReader.readFooter(
491-
conf, currentFile, SKIP_ROW_GROUPS)))
492-
} catch { case e: RuntimeException =>
493-
if (ignoreCorruptFiles) {
494-
logWarning(s"Skipped the footer in the corrupted file: $currentFile", e)
495-
None
496-
} else {
497-
throw new IOException(s"Could not read footer for file: $currentFile", e)
484+
try {
485+
parFiles.flatMap { currentFile =>
486+
try {
487+
// Skips row group information since we only need the schema.
488+
// ParquetFileReader.readFooter throws RuntimeException, instead of IOException,
489+
// when it can't read the footer.
490+
Some(new Footer(currentFile.getPath(),
491+
ParquetFileReader.readFooter(
492+
conf, currentFile, SKIP_ROW_GROUPS)))
493+
} catch { case e: RuntimeException =>
494+
if (ignoreCorruptFiles) {
495+
logWarning(s"Skipped the footer in the corrupted file: $currentFile", e)
496+
None
497+
} else {
498+
throw new IOException(s"Could not read footer for file: $currentFile", e)
499+
}
498500
}
499-
} finally {
500-
pool.shutdown()
501-
}
502-
}.seq
501+
}.seq
502+
} finally {
503+
pool.shutdown()
504+
}
503505
}
504506

505507
/**

0 commit comments

Comments
 (0)