Spark 处理小文件

不论是Hive还是Spark SQL在使用过程中都可能会遇到小文件过多的问题。小文件过多最直接的表现是任务执行时间长，查看Spark log会发现大量的数据移动的日志。我们可以查看log中展现的日志信息，去对应的路径下查看文件的大小和个数。

通过上述命令可以查看文件的个数以及大小。count查看出的文件大小单位是B，需要转换为MB。

在spark官方的推荐文档中，parquet格式的文件推荐大小是128MB，小于该大小的均可以称之为小文件，在实际的工作，往往小文件的大小仅仅为几KB，表现为，可能文件大小为几百MB，但是文件个数可能到达了几十万个。一般来说，我们可以通过简单相除获得文件的平均大小，如果文件数目不多，我们也可以通过下述命令获得每个文件的大小。

1.任务执行时间长

2.真实的文件大小独占一个数据存储块，存放到DataNode节点中。同时 DataNode一般默认存三份副本，以保障数据安全。同时该文件所存放的位置也写入到NameNode的内存中，如果有Secondary NameNode高可用节点，也可同时复制一份过去。NameNode的内存数据将会存放到硬盘中，如果HDFS发生重启，将产生较长时间的元数据从硬盘读到内存的过程。

3.不论在Hive还是在Spark中，每一个存储块都对应一个Map程序，一个Map呈现就需要一个JVM，启动一个JVM去读取或者写小文件是吃力不讨好的行为。在实际的生产中，为了更好的管理集群资源，一般会要求程序执行时限制Executor数量和每个Executor的核心数量，需要频繁创建Executor来读取写入。

5.影响磁盘寻址时间

小文件合并，本质上就是通过某种 *** 作，将一系列小文件合并成大文件。我们知道，以MapReduce为代表的大数据系统，都习惯用K-V键值对的形式来处理文件，最后文件落盘，也是一个reduce对应一个输出文件。所以直观上，我们可以减少reduce数量，达到减少文件数量的目的。

从Map到Reduce需要一个Shuffle过程，所以我们将小文件合并理解为通过一个Shuffle，合并小文件成一个大文件。基于这样的思想，我们的策略可以分为两类：一类是原来的计算已经有Shuffle了，那么我们可以认为控制输出文件的数量；二类是强制触发Shuffle，进行小文件合并。

1-设置参数 (一般用于Hive)

2-distribute by rand()

往动态分区插入数据时，在已经写好的SQL末尾加上distribute by rand()

该算子只是起到打散的效果，但是我们还要设置文件的大小，以免打散后仍然有小文件。

表示每个reduce的大小，Hive可以数据总量，得到reduce个数，假设hive认为会有10个reduce,那么，这里rand()则会为 x % 10

3-group by

我们知道，group by算子会触发Shuffle，因此只要我们设置好Shuffle时的文件个数就好，在Spark SQL中，我们可以设置partition个数，因为一个partition会对应一个文件。

上述的 *** 作，会触发shuffle，因此我们再设置partition个数。

则表示，shuffle后，只会产生10个partition.

4-repartition()

5-coalesce()

需要注意的是，4和5都是spark 2.4以及以后才会支持的。

1.输入:输入数据分为键/值对，由集群中的每个节点处理。2.映射函数:使用输入数据中的每个键/值对来调用用户定义的映射函数，以生成一组中间键/值对。3.Shuffle:将中间的键/值对分组，并将其发送到正确的节点。4.Reduce函数:将同一个键的中间值赋给reduce函数并聚合它们。5.输出:将最终的键/值对发送到输出文件。1)输入数据接口:InputFormat。默认的实现类是:TextInputFormat。TextInputFormat的功能逻辑是:一次读取一行文本，然后以该行的起始偏移量为键，将行内容作为值返回。CombineTextInputFormat可以将多个小文件合并成一个切片，提高处理效率。(2)逻辑处理接口:映射器用户根据业务需求实现三种方法:map() setup() cleanup()。(3)划分器划分HashPartitioner有一个默认实现，逻辑是根据key和numReduces的哈希值返回一个分区号；key.hashCode()&Integer。最大值% numReduces如果业务上有特殊需求，可以定制分区。(4)厘米平行排序当我们使用自定义对象作为输出的键时，必须实现WritableComparable接口，并在其中重写compareTo()方法。部分排序:每个最终输出文件的内部排序。全排序:对所有数据进行排序，通常只减少一次。二次排序:排序有两个条件。(5)合并者合并合并器合并可以提高程序执行的效率，减少IO传输。但使用时不得影响原业务处理结果。(6)逻辑处理接口:减速器用户根据业务需求实现三种方法:reduce() setup() cleanup()。(7)输出数据接口:OutputFormat默认的实现类是TextOutputFormat，功能逻辑是:为每个KV对输出一行到目标文本文件。用户还可以自定义输出格式。

使用Hadoop进行大数据运算，当数据量极其大时，那么对MapReduce性能的调优重要性不言而喻，尤其是Shuffle过程中的参数配置对作业的总执行时间影响特别大。下面总结一些和MapReduce相关的性能调优方法，主要从五个方面考虑：数据输入、Map阶段、Reduce阶段、Shuffle阶段和其他调优属性。

在执行MapReduce任务前，将小文件进行合并，大量的小文件会产生大量的map任务，增大map任务装载的次数，而任务的装载比较耗时，从而导致MapReduce运行速度较慢。因此我们采用CombineTextInputFormat来作为输入，解决输入端大量的小文件场景。

(1)减少溢写(spill)次数：通过调整io.sort.mb及sort.spill.percent参数值，增大触发spill的内存上限，减少spill次数，从而减少磁盘IO。

(2)减少合并(merge)次数：通过调整io.sort.factor参数，增大merge的文件数目，减少merge的次数，从而缩短mr处理时间。

(3)在map之后，不影响业务逻辑前提下，先进行combine处理，减少 I/O。

我们在上面提到的那些属性参数，都是位于mapred-default.xml文件中，这些属性参数的调优方式如表1所示。

表1 Map阶段调优属性

(1)合理设置map和reduce数：两个都不能设置太少，也不能设置太多。太少，会导致task等待，延长处理时间太多，会导致 map、reduce任务间竞争资源，造成处理超时等错误。

(2)设置map、reduce共存：调整slowstart.completedmaps参数，使map运行到一定程度后，reduce也开始运行，减少reduce的等待时间。

(3)规避使用reduce：因为reduce在用于连接数据集的时候将会产生大量的网络消耗。通过将MapReduce参数setNumReduceTasks设置为0来创建一个只有map的作业。

(4)合理设置reduce端的buffer：默认情况下，数据达到一个阈值的时候，buffer中的数据就会写入磁盘，然后reduce会从磁盘中获得所有的数据。也就是说，buffer和reduce是没有直接关联的，中间多一个写磁盘->读磁盘的过程，既然有这个弊端，那么就可以通过参数来配置，使得buffer中的一部分数据可以直接输送到reduce，从而减少IO开销。这样一来，设置buffer需要内存，读取数据需要内存，reduce计算也要内存，所以要根据作业的运行情况进行调整。

我们在上面提到的属性参数，都是位于mapred-default.xml文件中，这些属性参数的调优方式如表2所示。

表2 Reduce阶段的调优属性

Shuffle阶段的调优就是给Shuffle过程尽量多地提供内存空间，以防止出现内存溢出现象，可以由参数mapred.child.java.opts来设置，任务节点上的内存大小应尽量大。

我们在上面提到的属性参数，都是位于mapred-site.xml文件中，这些属性参数的调优方式如表3所示。

表3 shuffle阶段的调优属性

除此之外，MapReduce还有一些基本的资源属性的配置，这些配置的相关参数都位于mapred-default.xml文件中，我们可以合理配置这些属性提高MapReduce性能，表4列举了部分调优属性。

表4 MapReduce资源调优属性

转自： http://www.itheima.com/news/20201118/143040.html

---