Github user yhuai commented on a diff in the pull request:
https://github.com/apache/spark/pull/13444#discussion_r65823862
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/fileSourceInterfaces.scala
---
@@ -409,13 +409,24 @@ private[sql] object HadoopFsRelation extends Logging {
def listLeafFilesInParallel(
paths: Seq[Path],
hadoopConf: Configuration,
- sparkContext: SparkContext): mutable.LinkedHashSet[FileStatus] = {
+ sparkSession: SparkSession): mutable.LinkedHashSet[FileStatus] = {
+ assert(paths.size >=
sparkSession.sessionState.conf.parallelPartitionDiscoveryThreshold)
logInfo(s"Listing leaf files and directories in parallel under:
${paths.mkString(", ")}")
+ val sparkContext = sparkSession.sparkContext
+ val sqlConf = sparkSession.sessionState.conf
val serializableConfiguration = new
SerializableConfiguration(hadoopConf)
val serializedPaths = paths.map(_.toString)
- val fakeStatuses =
sparkContext.parallelize(serializedPaths).mapPartitions { paths =>
+ // Set the number of parallelism to prevent following file listing
from generating many tasks
+ // in case of large #defaultParallelism.
+ val numParallelism = Math.min(
+ paths.size / Math.max(sqlConf.parallelPartitionDiscoveryThreshold,
1) + 1,
+ sparkContext.defaultParallelism)
--- End diff --
I am not sure this `Math.min` can help if we have a small cluster (say,
defaultParallelism is 4). I think in general, we need to create more tasks than
`defaultParallelism` to help load balancing.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
