Spark 之 partitons

Listing leaf files and directories

分析其并行化

org.apache.spark.util.HadoopFSUtils

      sc.parallelize(paths, numParallelism).mapPartitions { pathsEachPartition =>val hadoopConf = serializableConfiguration.valuepathsEachPartition.map { path =>val leafFiles = listLeafFiles(path = path,hadoopConf = hadoopConf,filter = filter,contextOpt = None, // Can't execute parallel scans on workersignoreMissingFiles = ignoreMissingFiles,ignoreLocality = ignoreLocality,isRootPath = isRootLevel,parallelismThreshold = Int.MaxValue,parallelismMax = 0)(path, leafFiles)}}.collect()

    // 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, parallelismMax)

parallelismMax 最终由以下配置决定。

  val PARALLEL_PARTITION_DISCOVERY_PARALLELISM =buildConf("spark.sql.sources.parallelPartitionDiscovery.parallelism").doc("The number of parallelism to list a collection of path recursively, Set the " +"number to prevent file listing from generating too many tasks.").version("2.1.1").internal().intConf.createWithDefault(10000)