最近在專案中開展重構活動，對Map端記憶體儘量要省一些，當前的系統中Map端記憶體最高佔用大概3G左右（設定成2G時會導致Java Heap OOM）。雖然個人覺得佔用不算多，但是顯然這樣的結果想要試圖去說服一些對記憶體佔用非常挑剔的C++程式設計師們理由還是不夠，於是便通過一定的方式對記憶體的佔用進行了分析，刨根問底。

分析後的記憶體佔用

分析後的記憶體佔用峰值，即在我們的報表資料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個檔案，並將產生的檔案重新加入待合併列表中，重複以上過程以最終得到一個大檔案。