Set parallelism for the parallelize job in recursiveListDirs

delta-io · Sep 23, 2024 · 22a8433 · 22a8433
1 parent 1753cb5
commit 22a8433
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/util/DeltaFileOperations.scala b/spark/src/main/scala/org/apache/spark/sql/delta/util/DeltaFileOperations.scala
@@ -243,7 +243,10 @@ object DeltaFileOperations extends DeltaLogging {
     import org.apache.spark.sql.delta.implicits._
     if (subDirs.isEmpty) return spark.emptyDataset[SerializableFileStatus]
     val listParallelism = fileListingParallelism.getOrElse(spark.sparkContext.defaultParallelism)
-    val dirsAndFiles = spark.sparkContext.parallelize(subDirs).mapPartitions { dirs =>
+    val subDirsParallelism = subDirs.length.min(spark.sparkContext.defaultParallelism)
+    val dirsAndFiles = spark.sparkContext.parallelize(
+        subDirs,
+        subDirsParallelism).mapPartitions { dirs =>
       val logStore = LogStore(SparkEnv.get.conf, hadoopConf.value.value)
       listUsingLogStore(
         logStore,