微尘之患：探寻小文件问题的原理与对策
文章概览
背景
问题的引出
深入理解解决小文件问题的三板斧
输入小文件合并 - 揭开CombineHiveInputFormat类的神秘面纱
输出小文件合并 - 精通merge参数的奥义
输出小文件合并 - 另辟蹊径之distribute by
沙场秋点兵
创建文件总数过多
一次内容小时hive任务的优化
拓展：Spark中的小文件合并方案
参考文档
附录
施德伟，2023年加入去哪儿旅行，目前在大市场数仓团队担任数据开发工程师，主要负责新客运营和营销活动方向的数据开发、维护与优化工作。
1. 背景
1.1 小文件的定义
在大数据处理和存储中，"小文件" 通常是指文件大小远小于HDFS（分布式文件系统）中块（block）大小的文件。一般公司集群的block大小在128MB/256MB这二者的居多，因此，公司对小文件的大小没有一个统一的定义，通常小的不足1MB，大的甚至达到32MB或更大。
1.2 小文件的危害
小文件的危害有很多，这里主要从内存资源、计算资源和系统负载三个方面进行详细阐述：
① 内存资源的浪费：在HDFS中，每个存储对象block都需要元数据来描述其属性（如文件名、大小、权限、时间戳等）和位置信息，大概会占用150字节的空间。大量的小文件会导致元数据的数量急剧增加，这会占用大量的内存资源，并可能导致NameNode（HDFS的元数据管理节点）成为系统的瓶颈，影响整个集群的性能。
下图是1个200M的文件在有小文件和没有小文件2种场景下占用的内存空间对比情况：
' fill='%23FFFFFF'%3E%3Crect x='249' y='126' width='1' height='1'%3E%3C/rect%3E%3C/g%3E%3C/g%3E%3C/svg%3E)
无小文件场景下：有2个block块，每个block块有3个副本，此时占用NameNode的内存为900字节；
有大量的小文件场景下：假设有200个1M的block，每个block块有3个副本，此时占用的NameNode内存为90000字节，相比正常情况下多出了100倍的存储空间。
② 计算资源的浪费：基于 HDFS 文件系统的计算，block 块是最小粒度的数据处理单元，当计算任务读取的数据中存在大量小文件时，会启动大量Map任务，其中Map任务的启动又是非常消耗性能的操作，但是由于文件太小，启动后执行很短时间就停止了，导致任务启动的时间可能还要大于任务执行的时间，极大浪费了计算资源。
③ 系统负载增加：短时间读写/创建大量小文件，会造成NameNode节点的瞬时请求量过大，导致处理不及时，进而增加NameNode的响应时间，影响集群性能。
1.3 小文件产生的途径
日常生产中 HDFS 上小文件的产生是一个很正常的事情，有些甚至不可避免。下面是产生小文件的常见途径：
数据源本身就包含有大量的小文件，或数据源经过计算过滤后变成小文件；
流式数据，如 kafka、sparkstreaming、flink 等流式增量文件，小窗口文件，如几分钟一次等；
采用
动态分区也会产生很多小文件，特别是在过度分区的场景下。
1.4 如何解决
既然小文件的危害如此之大，而且它的产生也似乎不可避免，那如何解决小文件问题也显得至关重要。通过1.3节可以知道小文件基本都是在分布式计算场景下产生的，如果能够在分布式计算的同时避免小文件的产生，那问题也会迎刃而解。公司现在使用的主流分布式计算框架有Hive（底层使用Mapreduce计算）和Spark。
① 在Hive中，官方提供了一些官方参数来解决小文件问题，其主要思想是将小文件进行合并，将原始较小的多个小文件合并成一个大文件。Hive中解决小文件问题的三板斧分别是：（第三章详细阐述）
对输入小文件进行合并 - 通过配置CombineHiveInputFormat等参数实现
对输出小文件进行合并 - 通过配置merge参数实现
对输出小文件进行合并 - 通过在代码中加入distribute by
② 而在Spark中，并没有像Hive那样官方参数来直接解决小文件问题，但解决小文件的底层原理是相通的，下面是Spark中二种主流的解决方案：（第五章详细阐述）
使用Spark扩展功能实现
自定义一个包含小文件合并功能的commitProtocolClass协议类
2. 问题的引出
刚来公司那会，仅靠"CV大法"就"解决了"我的小文件问题。通过复制粘贴同事们设置好的参数，小文件问题就再也没关注过了... 但实际上对这些设置好的参数只是一知半解，一遇到问题也不知道如何去调，别人让调大就调大，别人让调小就调小，调的过程中只会觉得别人说的有道理！直到有一天我跑了个任务，明明设置了最大切片大小（maxSplitSize）为128M，可是最后生成的文件竟然接近200M？还有一天我CV好了合并的参数，可是最后生成的文件却又远小于我设置的参数？等等... 于是乎决定好好研究一番，做到知其然，知其所以然。最后通过阅读源码和同事导师的帮助下总结出了这篇文章。
3. 深入理解解决小文件问题的三板斧
3.1 输入小文件合并 - 揭开CombineHiveInputFormat类的神秘面纱
首先介绍的是合并小文件的第一板斧 - CombineHiveInputFormat类。当Hive读取的数据源中有很多小文件时，这个类可以在读取数据的同时对这些小文件进行合并。在Hive中，这个功能是默认开启的，但需要通过设置以下参数来控制合并的大小，设置不当可能会导致参数设置无效：
通过上方参数可以看到，CombineHiveInputFormat类是发挥作用的主心骨，那它是何方神圣呢，通过图1的继承类图关系可以看出，它合并小文件的核心方法并不是自己实现的，而是调用了hadoop中的CombineFileInputFormat类的getSplits方法来实现的（在图中已用红色五角星标注）。
图1. InputFormat家族类图关系
在说到CombineFileInputFormat的时候，我们就不得不先说下FileInputFormat了，FileInputFormat和CombineFileInputFormat都是抽象类，他们各自实现了getSplits(获取切片)方法，只不过CombineFileInputFormat的getSplits方法拥有合并小文件的功能，也可以说它的诞生就是来解决FileInputFormat的这个痛点的。
既然说到这了，不妨回顾下FileInputFormat是如何对block进行切分的，在学习mapreduce的wordcount案例的时候，我们应该都使用过TextFileInputFormat类，这个类就是FileInputFormat的实现类，它就是调用了FileInputFormat的getSplits方法来实现block的切分的，下面是getSplits方法的部分源码，需要注意的是FileInputFormat只会对数据进行逻辑切分，而且切分会有buff，1.1倍才切，这个设置是为了防止生成特别小的切片。
回顾完FileInputFormat，我们再来看今天的主角CombineFileInputFormat类，CombineFileInputFormat类不仅会对数据进行逻辑切分，在切分后还会再进行逻辑合并，所以这个过程可以分为2个步骤来进行讲解：
3.1.1逻辑切分
如图2，假设我们有10个文件分别为F1、F2...F10，其中F为File的缩写，这10个文件中有几个较小的文件，如F10仅有5M大小，它们分别散落在2个机架（Rack）中的5台机器节点（Node）上，我们将以这10个文件来讲述CombineFileInputFormat类整个生命周期是如何工作的。大致生命周期如下：
逻辑切分会将文件按照参数mapred.max.split.size（默认为128M）的大小切分成一个个的逻辑block块，它的作用是防止文件过大，致使文件并行计算的效率降低。这里的切分并没有多余的buff，只要大于maxSplitSize就会进行切分。图2展示了这个切分过程（File → Block）：
图2. CombineFileInputFormat的逻辑切分过程
注意：这里的切分，并不是物理上的切分，只是逻辑上的切分，简单的说，就是记录切分的起始位置和长度等元信息。
看完切分结果后，可能会产生这样一个疑问，File9文件大小为142M，为什么会被切分成了2个同样大小的71M？而不是128M+14M？通过下面的源码可以揭秘这个过程：
在切分的过程中，CombineFileInputFormat还会将block的Node节点信息和机架Rack信息都存储起来，以便于后续进行逻辑合并工作，具体的存储方式是通过2个HashMap来存储的：
注意：在源码中，逻辑切分后产生的叫block块（和HDFS的block不是一个意思），但实际上也可以理解成切片，因为它也是切分后的结果，猜测是为了和下个步骤-逻辑合并中的切片作区分。
3.1.2 逻辑合并
通过3.1.1节中的图2可知，我们从10个File，切分成了15个Block，我们继续观察这15个Block是如何合并成Split的，图中的b是block的简写，s是split的简写。
逻辑合并分为了3个阶段，分别为节点内部的合并、机架内部的合并、机架之间的合并。
阶段1：节点内部的合并
该阶段是Node节点内部的block合并过程，会循环遍历每个Node节点中的block进行合并，满足条件的会合并成split，不满足条件的会放入池中等待下个阶段进行处理，图3展示了这个过程：
图3的合并过程示例是在下面3个参数的基础上进行的：
set mapred.max.split.size = 128M
set mapred.min.split.size.per.node = 32M
set mapred.min.split.size.per.rack = 64M
该阶段合并成功的split切片有s1、s2、s3、s4、s5、s6
节点内合并的过程源码如下，源码比较晦涩，为了便于理解，可以结合图3和图4来理解这个过程：
图4. 节点(Node)内合并的流程图
下面以Node01和Node03两个具有代表性的节点来讲解这个合并过程：
在Node01内部，循环遍历其所有block块：
由于b1 = 128M，满足 >= maxSplitSize（128M），则合成第一个切片s1，此时跳到下个Node → Node02的循环中...；
然而Node01并未完成所有合并工作，仍会回到Node01的循环中，此时b2 = 86M，不满足 >= maxSplitSize（128M），则累加下一个block大小，此时b2 + b3 = 172M >= maxSplitSize（128M），则合成切片s6；
同理，b4 + b5 = 110M，不满足 >= maxSplitSize（128M），此时节点内的block全部遍历结束，跳出循环；
随后进入下一个判断，b4 + b5 = 110M，满足 >= minSizeNode（32M），但并不满足当前节点没有合并过切片这个判断，所以将b4和b5放入池中，等待下个rack阶段处理，同时该节点也完成了所有的合并工作。
在Node03内部，循环遍历其所有block块：
由于b9 = 60M，不满足 >= maxSplitSize（128M），此时节点内的block全部遍历结束，跳出循环；
随后进入下一个判断，b9 = 60M，满足 >= minSizeNode（32M），且满足当前节点没有合并过任何切片这个判断，所以将b9合并成切片s3。
备注：在阶段1结束后，被放入池中待下阶段处理的block有：b4、b5、b8、b12和b15。
阶段2：机架内部的合并
该阶段是Rack机架内部的block合并过程，会循环遍历每个Rack节点中的block进行合并，满足条件的会合并成split，不满足条件的会放入池中等待下个阶段进行处理，图5展示了这个过程：
该阶段合并成功的切片有：s7
节点内合并的过程源码如下，另外为了便于理解，图6的流程图可视化了这个过程：
图6. 机架(Rack)内合并的流程图
下面演示Rack01中的切片s7的合并过程：
由上文可知，b4、b5、b8、b12和b15五个数据块在阶段1结束后都被放入了池中待本阶段处理；
在Rack01内部循环中，b4+b5+b8 = 130M >= maxSplitSize（128M），则合并b4、b5、b8为切片s7，从源码中可知，此时会直接跳出Rack01的循环，进入到Rack02的循环中；
在Rack02内部循环中，由于b15 = 5M不满足 >= maxSplitSize（128M），且不满足 >= maxSizeRack（64M），所以b15会被放入到overflow集合中待阶段3进行处理，需要注意的是Rack01中的b12也在此时被同时放入overflow集合中。
阶段3：机架之间的合并
该阶段是Rack机架之间的block合并过程，会循环Rack里剩余的block进行最后的合并，图7展示了这个过程：
该阶段合并成功的切片有：s8
机架之间的合并的过程源码如下，另外为了便于理解，图8的流程图可视化了这个过程：
图8. 机架(Rack)之间合并的流程图
下面演示s8的合并过程：
开始遍历overflow集合中的未处理block，b12+b15 = 15M不满足 >= maxSplitSize（128M），不进行合并，遍历结束；
进入下一个判断，是否还有剩余的block未处理，满足条件，所以将b13和b16合成切片s8；
到此，完成合并的所有过程。
3.1.3 总结
通过上面的过程，我们可以看到从一开始的10个File文件（有几个小文件），合并成最后的8个Split（每个文件相对均匀），在这个过程中即避免了文件过大造成的并行度低问题，又解决了小文件的问题。最后8个Split会生成8个并行的Map计算任务，且每个Map任务重处理的数据量相对均匀（如果没有CombineFileInputFormat类的合并，将会生成10个Map任务，且其中的几个任务读取的数据量过小，甚至会造成任务启动的时间还要大于任务执行的时间，造成计算资源的极大浪费）。
备注：该案例只是为了尽可能展示出合并过程中的所有情况，也无法绘制出过多过小的小文件图示，所以合并的效果看上去并不是特别明显，正式环境中小文件的数量可能达到几十，甚至几百个，那时合并的效果就会显得异常明显。
3.2 输出小文件合并 - 精通merge参数的奥义
3.2.1 输出小文件的烦恼
虽然通过设置max.split.size等参数对输入的小文件进行了合并，但输出的时候仍然可能产生大量小文件，因为在SQL中往往还会对输入的数据进行过滤，不限于where条件的筛选和字段的筛选，导致本来map任务读取的数据有一二百兆，到输出的时候数据量骤降，有时甚至不到1M。
下面就是个产生了大量输出小文件的案例：
案例SQL如下：
执行日志如下：
通过日志可以看到:
HDFS读取的数据量有112985074713，一共启动了1029个map任务，平均每个map读取的数据量为：112985074713/1024/1024/1029 = 105M左右；
HDFS输出的数据量有935980245，一共启动了2210个reduce任务（根据map的输入数据量估算出来的），所以最终平均输出的文件大小只有935980245/1024/1024/2210 = 0.4M，输出了2210个小文件。
3.2.2 merge参数的登场
所以在输出的时候我们也希望拥有小文件合并的功能，这个时候第二板斧 - merge参数就闪亮登场了：
它会判断输出目录下的文件平均大小是否小于 hive.merge.smallfiles.avgsize，若小于，则单独再启动一个map only任务来进行小文件的合并，下面是该逻辑的源码：
这个单独的map only任务的底层其实还是通过3.1章节中介绍的CombineFileInputFormat类来完成文件的合并的（又执行了一遍这个逻辑），只不过它在执行前会重置MaxSplitSize、MinSplitSizePerNode、MinSplitSizePerRack等参数。重置的逻辑如下：
通过设置merge参数后，再次执行了这个任务。
执行日志如下：
通过日志可以看到：
在job01执行完之后，再次启动了一个job02任务，这个任务是个map only任务，启动了14个map任务，合并了之前的2210的小文件为14个大文件，解决了输出小文件问题。
3.2.3 版本的bug
需要注意的是，公司使用的hive-2.1.1版本在重置MaxSplitSize、MinSplitSizePerNode等参数时会有个bug：
当目标表的写入格式是rcfile或者orcfile时，重置参数不会生效。在启动的合并任务中，MaxSplitSize、MinSplitSizePerNode等参数值并没有重置成功，取得仍然是mapred.max.split.size、mapred.min.split.size.per.node等参数中的值，该bug在hive-2.2.0中官方才进行了修复。但公司部分业务线的数仓团队通过修改hive-2.1.1的源码对这个bug也进行了个性化的修复。
3.3 输出小文件合并 - 另辟蹊径之distribute by
3.3.1 动态分区产生的隐患
动态分区也是产生小文件的元凶之一，特别是在分区粒度过细的情况下：
当没有动态分区时，一般最终生成的文件个数和最后一个Job的map/reduce task个数是一样的；
当设置了动态分区的情况下，每个分区内的文件个数和最后一个Job的map/reduce task个数有关，如果在最后的task数量是4，那么每个分区内大概率是4个文件（除非有个task内没有本分区的数据）。
所以假设insert语句有100个dt分区的时候，那么最后产生的文件数可能接近原始文件数的100倍。但动态分区在我们开发的过程中经常是不可避免的，那有什么办法能避免这么多文件的产生呢，这时候，distribute by就隆重登场了，distribute by可以让拥有相同key的数据分发到同一个reduce task中，那distribute by dt就可以让相同dt的数据分发到同一个reducer中，这样每个reduce task就只会生成一个文件了。下面通过一个简单的实战案例和流程图解来更好的了解这个过程：
图9. 未使用distribute by流程图
可以看到最后的reduce阶段有2个reduce task，分别向dt=2024-01-01和dt=2024-01-02这两个分区写入了2个文件，一共生成了4个文件。
3.3.2 善用distribute by
图10. 使用distribute by的流程图
可以看到，通过distribute by dt重新启动了一个Job02，按照key对数据进行了重新分发，让拥有相同dt的数据分发到同一个reduce中，最后2个reduce分别向各自的分区只写了一个文件，总共只生成了2个文件。
备注：当在一个map only的Job任务上加入了distribute by时，并不会再重新启动一个Job，只会将原本的map only任务变成一个map reduce任务。
4. 沙场秋点兵
4.1 创建文件总数过多
在执行一次插入动态分区的Hive SQL任务时，任务失败了：
通过查看日志中的错误信息发现，是由于创建的文件数目过多导致的：
因为Hive对文件创建的总数是有限制的，这个限制取决于参数：hive.exec.max.created.files，该参数的默认值是100000，也就是在reduce阶段，hive发现生成的文件数超过了10万，而这种文件量级被认为是不健康的，所以hive直接kill了当前job。
这时有2个解决方案：
解决方案1：修改参数默认值
既然最大创建的文件数是有这个参数决定的：hive.exec.max.created.files，那我们第一个想到的可能就是提高这个参数的值，第二次执行时将这个参数的值设置成了150000，任务也正常执行完成了。
虽然任务不再报错了，但随之带来2个问题：
创建的文件数量过多；
hive.exec.max.created.files这个参数无法动态设置，就算设置成了150000，下次可能还会超过这个阈值，任务仍然会报错。
首先分析下第一个问题，创建的文件数量超过默认值10万确实已经很多了，hive官方设置的默认值也有它的考究之处。但我们眼尖的同学应该也能发现，实际上最后生成的文件数并不一定有10万这么多，因为merge参数在最后可能还会发挥作用，重新启动合并任务来合并这些小文件，但是文件数过多其实对于合并任务来说也会增加开销。
另外第二个问题才是重点，我们无法预测后期执行的任务创建的文件数阈值，这个参数也不会动态去调整，就算设置成15万，后期也会有再次超过阈值的风险。那有没有其他的解决方案呢？
解决方案2：使用distribute by
到这里，眼尖的同学可能就会想到distribute by了，它的加入可以让数据进行重新分发，在一定程度上可以减少每个分区中的文件数。我们在sql的结尾加上了distribute by dt后再次执行了任务，任务也正常执行完成了。
这里的源码就不再阐述了，这个方案也完美的解决了方案1中的2个问题，但是如果某个分区的数据量异常多，这个方案会出现数据倾斜的问题。
可以通过执行日志看到，最后一个job不再是map only任务，而是一个map reduce任务。
4.2 一次内容小时hive任务的优化
该问题是同事在工作中遇到的一个问题，由于内部参考文章无法访问，将其整理到此小节。
insert overwrite table t partition(dt) select .... distribute by dt 。在一次简单查询后直接insert的语句竟然运行了15分钟，通过查看mr运行情况发现有个别reduce运行很慢，经过排查发现select的查询其实只有两天的数据，只不过每天的数据较大，大约在7G左右，而写入的数据每天大约在1G左右，reduce的个数是根据默认的数据量推测的，大概有10几个，其实只有两个有数据，也就是这两个reduce任务慢；那有没有一种方法既可以运行的快又可以避免小文件的问题呢？
① sql语句去除distribute by dt让job没有reduce
② 设置如下参数让hive自动合并map输出的小文件
③ 由于这个sql的select语句中大量使用了正则匹配，所以为了加快map的速度可以设置如下参数增加map的数量
注意：select查询的表是text存储格式，文件并进行了gz压缩，由于gz不支持分割但是支持合并，故此处设置split大小为location下最大文件的大小，即128M；这样虽然没有大幅提高map的速度，但是减少了map数，从而节省了资源。
④ 完整的参数信息如下：
经过这么设置后sql会有两个job，两个job都是只有map没有reduce，第一个job是执行自己的sql逻辑，第二个job是合并小文件；运行速度从之前的15min缩减到6min，目标表的文件由之前的1个1G多点的文件变成1个接近1G和1个几百兆的两个文件，即提升了运行速度又避免了小文件的产生，完美解决了问题。
需要注意的是：参数 hive.optimize.sort.dynamic.partition 默认值的是true，也就是在动态分区的时候默认就会按照分区字段进行排序，和对分区字段加了distribute by语义一样。假如此参数设置为true，那么如果sql里面没有指明以分区字段进行shuffle的话会自动以其为key，让原本只有只有map的任务产生reduce，本来以其他字段shuffle后应该结束的，结果又新起了一个job以分区字段为key又产生一次shuffle，极其不友好，所以这里将其设为false。
5. 拓展：Spark中的小文件合并方案
在Spark中，没有像hive那样的官方参数来解决小文件问题，但小文件合并的原理是相通的，现在主流的解决方案有以下2种：
5.1 通过Spark扩展功能
通过Spark的扩展功能来实现：
首先介绍下什么是Spark的扩展功能。如图11所示，Spark SQL语句在执行前会进行逻辑计划的优化和物理计划的优化，继而生成可执行的物理计划。在这个过程中我们可以获取到Spark任务的一些执行信息，通过这些信息用户可以实现一些自定义的扩展能力。
Spark约在四个阶段提供了扩展点，分别是SQL Parse阶段、Analyze阶段、Optimizer阶段、Planning Strategies阶段。
图11. catalyst优化过程
我们可以通过实现自定义的扩展类来实现小文件合并的功能，具体可通过set spark.sql.extensions=xxx参数来完成，其中xxx是自定义的扩展类，例如可以使用网易Kyuubi提供的org.apache.kyuubi.sql.KyuubiSparkSQLExtension类就提供了该功能，Kyuubi 可以通过添加额外的 shuffle 来合并小文件，其实现原理与Hive也较为相似。
备注：公司目前的Spark小文件合并功能也是选择的该方案。
5.2 重写commitProtocolClass协议类
通过使用set spark.sql.sources.commitProtocolClass=xxx 的方式引入包含合并小文件功能的 CommitProtocol 协议类。
通过源码可以看到Spark默认的spark.sql.sources.commitProtocolClass=org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol，我们只要自定义一个包含小文件合并功能的commitProtocolClass协议类，并通过参数设置就可直接使用。
那commitProtocolClass是如何起作用的呢。Spark任务在执行过程中，将数据写入到HDFS时，为了解决数据一致性的问题，其中有两个非常关键的操作，可结合图12理解：
executor端的task任务执行commitTask方法，将数据文件从task临时目录转移到Job临时目录；
driver端执行commitJob方法，将各个task任务提交的数据文件，从Job临时目录转移到Job的最终目录。
图12. FileOutputCommitter 文件提交机制[1]
而commitProtocolClass实现小文件合并的基本原理是：在executor端，各个task任务执行完commitTask方法写入数据后，通过重写的协议类，读取到所有小文件，然后在各分区内对小文件进行分组合并，最后driver端在执行commitJob方法时，将合并后的文件移动到Job的最终目录。[1]
6.参考文档
[1] Spark 小文件合并功能在 AWS S3 上的应用与实践
[2] Auxiliary Optimization Rules
[3] https://issues.apache.org/jira/browse/HIVE-15178
7. 附录
节点Node内部的合并过程动图：
刷新 
