本文_SPARK中会对Scan的小文件做合并到一个Task去处理么？

篇首语：本文由编程笔记#小编为大家整理，主要介绍了SPARK中 会对Scan的小文件做合并到一个Task去处理么？相关的知识，希望对你有一定的参考价值。

背景

本文基于SPARK 3.1.2
在之前查看SQL物理计划的时候&＃xff0c;发现一个很奇怪的现象&＃xff0c;文件的个数很多&＃xff0c;但是启动的Task却很少。

结论

SPARK在scan文件的时候&＃xff0c;会把小文件合并到一个Task上去处理。

分析

这里的SQL很简单&＃xff1a;就是select col from table语句我们直接查看对应的计划&＃xff1a;

可以看到对于有50000多个文件的source&＃xff0c;最终却只有6000多个任务运行。
我们直接看对应的代码FileSourceScanExec实现&＃xff1a;

val splitFiles &＃61; selectedPartitions.flatMap partition &＃61;>
partition.files.flatMap file &＃61;>
// getPath() is very expensive so we only want to call it once in this block:
val filePath &＃61; file.getPath
val isSplitable &＃61; relation.fileFormat.isSplitable(
relation.sparkSession, relation.options, filePath)
PartitionedFileUtil.splitFiles(
sparkSession &＃61; relation.sparkSession,
file &＃61; file,
filePath &＃61; filePath,
isSplitable &＃61; isSplitable,
maxSplitBytes &＃61; maxSplitBytes,
partitionValues &＃61; partition.values
)

.sortBy(_.length)(implicitly[Ordering[Long]].reverse)
val partitions &＃61;
FilePartition.getFilePartitions(relation.sparkSession, splitFiles, maxSplitBytes)
new FileScanRDD(fsRelation.sparkSession, readFile, partitions)


PartitionedFileUtil.splitFiles就是对每个文件进行遍历&＃xff0c;如果一个文件超过了maxSplitBytes,这个可以参考Spark-读取Parquet-为什么task数量会多于Row Group的数量,就进行切分&＃xff0c;否则就直接返回整个文件&＃xff0c;
关键的在FilePartition.getFilePartitions(relation.sparkSession, splitFiles, maxSplitBytes):

...
partitionedFiles.foreach file &＃61;>
if (currentSize &＃43; file.length > maxSplitBytes)
closePartition()

// Add the given file to the current partition.
currentSize &＃43;&＃61; file.length &＃43; openCostInBytes
currentFiles &＃43;&＃61; file

...


她会根据maxSplitBytes来判断&＃xff0c;如果文件小于该阈值&＃xff0c;就会放到同一个FilePartition中&＃xff0c;从而让一个Task去处理&＃xff0c;这样就会出现了图上所展示的小文件很多但是Task缺比较小的现象。