项目中Map端内存占用的分析

 

最近在项目中开展重构活动，对Map端内存尽量要省一些，当前的系统中Map端内存最高占用大概3G左右（设置成2G时会导致Java Heap OOM）。虽然个人觉得占用不算多，但是显然这样的结果想要试图去说服一些对内存占用非常挑剔的C++程序员们理由还是不够，于是便通过一定的方式对内存的占用进行了分析，刨根问底。

 

关于运行时内存占用可以参考文章：http://brandnewuser.iteye.com/blog/2113828， 这里采用的是简单的方式，通过反射将内存MemoryCounter的方式计算的内存占用。

 

分析后的内存占用

 

分析后的内存占用峰值，即在我们的报表数据dump之前，就是属于内存占用的峰值。由于报表对象占用的数据非常大，大概1G左右，而且Map的输出中介结果Value格式是BytesWritable，是通过Java序列化方式dump出来的，因此当时的byte[]非常大（甚至可能超出1G）。于是，便采用了拆分报表输出的方式，这样便可以节省一定的内存空间。

 

在分析内存的占用过程中，尽量查找一些不太合理的内存占用，于是我们便查找到其中的一个类，在初始化后，占用500M的内存，这就是其中使用到的org.apache.hadoop.fs.FSDataInputStream，为什么这个类的对象会占用500M内存？

 

代码中初始化org.apache.hadoop.fs.FSDataInputStream后，就立刻占用500M的内存，那么是否这个类我们使用的方式不对？经过我们单独写了一个应用程序的测试（读取的HDFS与上面的环境一致），其内存占用并没有达到哪怕是1M。

 

org.apache.hadoop.fs.FSDataInputStream接收一个InputStream参数，经过打印其具体实现类型，发现为：org.apache.hadoop.hdfs.DFSInputStream，初始化这个类时，内存占用就已经上去了。

     

于是在MemoryCounter中加入日志，将对象size大于100M的对象打印出来，发现有一个byte[]长度为536870912占用内存！

Exceed 100M object[Array]:
byte Exceed 100M object[Array]:
[B Array length: 536870912

 

 

初步计算了一下byte[]长度为536870912，估计为2的29次幂，根据条件打印计算出来的结果，发现：

Start estimate: org.apache.hadoop.mapred.MapTask$MapOutputBuffer: 
Start estimate: org.apache.hadoop.util.QuickSort: 
Start estimate: [B: 
Start estimate: [B: 
Exceed 100M object[Array]: [B

于是恍然大悟，是MapOutputBuffer，然后查看了任务配置的参数，果然

mapreduce.task.io.sort.mb＝512

 

缓冲区设置占用了512M，也就是2的29次幂，这是需要占用Map端的Java Heap内存，下面就是对这部分的一个学习和回顾。

 

Map Collect 过程分析

 

待每次map函数处理完一对key/value，并产生新的key/value后，就会调用OutputCollector.collect函数输出结果，函数内部首先会调用Partitioner.getPartition()获取纪录的分区号，将<key, value, partition>传递给MapOutputBuffer.collect函数作进一步的处理。

 

MapOutputBuffer内部使用了一个缓冲区暂时存储用户输出数据，当缓冲区的数据达到一定阈值的时候，再将缓冲区的数据写到磁盘上。缓冲区采用的是环形内存缓冲区保存数据(大小为mapreduce.task.io.sort.mb)，当达到一定数值(mapreduce.map.sort.spill.percent)后，由线程SpillThread将数据写到一个临时文件中，当所有的数据都处理完成后，对所有的临时文件进行一次合并生成一个最终文件。环形缓冲区使得Map Task的Collect阶段和Spill阶段可以并行地进行。

 

MapOutputBuffer采用了两级索引结构，涉及到3个环形内存缓冲区，三个缓冲区占用内存的总空间为mapreduce.task.io.sort.mb。

 

 

1. kvoffsets, 偏移量索引数组，用于保存key/value信息在位置索引kvindices中的偏移量，一对key/value需要占用kvoffsets的1个int大小，数组kvindices的3个int大小
2. kvindices, 位置索引数组，用于保存key/value值在数据缓冲区kvbuffer中的起始位置
3. kvbuffer，数据缓冲区，用于保存实际的key/value值，默认情况下可以最多使用整个缓冲区的95%

 

以上几个缓冲区读写均采用了典型的生产者消费者模型，MapOutputBuffer的collect方法和MapOutputBuffer的write方法是生产者，spillThread线程是消费者，通过可重入互斥锁的条件变量来完成的。

 

Spill过程是由SpillThread线程完成的，SpillThread线程实际上是缓冲区kvbuffer的消费者，调用sortAndSpill方法将环形缓冲区kvbuffer中的数据写到磁盘上，函数的工作原理如下：

 

1. 利用快速排序算法对缓冲区kvbuffer的数据进行排序，先按照partition进行排序，然后按照key进行排序。经过排序后，数据以分区为单位聚集在一起，同一分区内的所有数据按照key有序；
2. 按照分区编号由小到大依次将每个分区中的数据写入任务工作目录下的临时文件output/spillN.out（N为spill的次数），如果用户设置了Combiner，写入文件之前可能会对每个分区的数据进行一次数据聚集操作；
3. 将分区数据的元信息写到内存索引数据结构SpillRecord中，每个分区的元信息包括在临时文件的偏移量，压缩前数据大小和压缩后数据大小，如果内存中的索引大小超过1M，将内存索引写到索引文件中output/spillN.out.index中。

 

当所有数据处理完成之后，Map Task会将所有的临时文件合并成一个大的文件，保存到output/file.out.index中。在进行文件合并的过程中，以分区为单位，对于某个分区，采用多轮递归合并的方式：每轮合并io.sort.factor个文件，并将产生的文件重新加入待合并列表中，重复以上过程以最终得到一个大文件。