Hadoop效能調優全面總結
一、 Hadoop概述
隨著企業要處理的資料量越來越大，MapReduce思想越來越受到重視。Hadoop是MapReduce的一個開源實現，由於其良好的擴充套件性和容錯性，已得到越來越廣泛的應用。
Hadoop實現了一個分散式檔案系統（Hadoop Distributed File System），簡HDFS。HDFS有高容錯性的特點，並且設計用來部署在低廉的（low-cost）硬體上；而且它提供高吞吐量（high throughput）來訪問應用程式的資料，適合那些有著超大資料集（large data set）的應用程式。HDFS放寬了（relax）POSIX的要求，可以以流的形式訪問（streaming access）檔案系統中的資料。
Hadoop的框架最核心的設計就是：HDFS和MapReduce。HDFS為海量的資料提供了儲存，則MapReduce為海量的資料提供了計算。Hadoop的架構圖如下所示：
Hadoop作為一個基礎大資料處理平臺，雖然其應用價值已得到大家認可，但仍存在很多問題，以下是主要幾個：
1、Namenode/jobtracker單點故障。
Hadoop採用的是master/slaves架構，該架構管理起來比較簡單，但存在致命的單點故障和空間容量不足等缺點，這已經嚴重影響了Hadoop的可擴充套件性。
2、HDFS小檔案問題。
在HDFS中，任何block，檔案或者目錄在記憶體中均以物件的形式儲存，每個物件約佔150byte，如果有1000 0000個小檔案，每個檔案佔用一個block，則namenode需要2G空間。如果儲存1億個檔案，則namenode需要20G空間。這樣namenode記憶體容量嚴重製約了叢集的擴充套件。
3、jobtracker同時進行監控和排程，負載過大。
為了解決該問題，yahoo已經開始著手設計下一代Hadoop MapReduce（見參考資料1）。他們的主要思路是將監控和排程分離，獨立出一個專門的元件進行監控，而jobtracker只負責總體排程，至於區域性排程，交給作業所在的client。
4、資料處理效能。
很多實驗表明，其處理效能有很大的提升空間。Hadoop類似於資料庫，可能需要專門的優化工程師根據實際的應用需要對Hadoop進行調優，有人稱之為“Hadoop Performance Optimization” (HPO)。
為了提高其資料效能，很多人開始優化Hadoop。總結看來，對於Hadoop，當前主要有幾個優化思路：
(1)作業系統調優
Hadoop的執行環境，硬體配置起得至關重要的作用，硬體的引數配置對效能影響非常大，在部署Hadoop時，合理的硬體選擇是一種優化思路。
(2)從應用程式角度進行優化。
由於mapreduce是迭代逐行解析資料檔案的，怎樣在迭代的情況下，編寫高效率的應用程式，是一種優化思路。
(3)對Hadoop引數進行調優。
當前hadoop系統有190多個配置引數，怎樣調整這些引數，使hadoop作業執行儘可能的快，也是一種優化思路。
(4)從系統實現角度進行優化。
這種優化難度是最大的，它是從hadoop實現機制角度，發現當前Hadoop設計和實現上的缺點，然後進行原始碼級地修改。該方法雖難度大，但往往效果明顯。
二、 執行環境
原則一： 主節點可靠性要好於從節點。
原則二：多路多核，高頻率cpu、大記憶體。
比如，NameNode節點中100萬檔案的元資料要消耗800M記憶體，記憶體決定了叢集儲存檔案數的總量，ResourceManager同時執行的作業會消耗一定的記憶體。
DataNode的記憶體需要根據cpu的虛擬核數(vcore) 進行配比，CPU的vcore數計算公式為=cpu個數 * 單cpu核數* HT（超執行緒）
記憶體容量大小 = vcore數 * 2GB(至少2GB)
原則三： 根據資料量確定叢集規模
一天增加10GB, 365天，原資料1TB，replacation=3， 1.5個mapreduce 計算完儲存的資料，規劃容量
（1TB + 10GB*365）*3*1.5 =20.53TB
如果一臺datanode的儲存空間為2TB，21/2=11，總節點為 = 11+2 =13
還要考慮作業並不是均勻分佈的， 有可能會傾斜到某一個時間段，需要預留資源。
原則四： 不要讓網路I/O 成為瓶頸
hadoop 作業通常是 I/O密集型而非計算密集型， 瓶頸通常集中出現在I/O上， 計算能力可以通過增加新節點進行線性擴充套件，要注意網路設別處理能力。
三、 作業系統調優
(1) 避免使用swap 分割槽 將hadoop守護程序的資料交換到硬碟的行為可能會導致操作超時。
(2) 調整記憶體分配策略
操縱系統核心根據vm.oversommit_memory的值來決定分配策略，並且通過vm.overcommit_ratio的值來設定超過實體記憶體的比例。
(3) 修改net.core.somaxconn引數
該引數表示socker監聽backlog的上限，預設為128,socker的伺服器會一次性處理backlog中的所有請求，hadoop的ipc.server.listen.queue.size引數和linux的net.core.somaxconn
引數控制了監聽佇列的長度，需要調大。
(4) 增大同時開啟檔案描述符的上限
對核心來說，所有開啟的檔案都通過檔案描述符引用，檔案描述符是一個非負整數，hadoop的作業經常會讀寫大量檔案，需要增大同時開啟檔案描述符的上限。
(5)選擇合適的檔案系統，並禁用檔案的訪問時間
ext4 xfs ,檔案訪問時間可以讓使用者知道那些檔案近期被檢視或修改， 但對hdfs來說，獲取某個檔案的某個塊 被修改過，沒有意義，可以禁用。
(6)關閉THP (transparent Huge Pages)
THP 是一個使管理 Huge Pages自動化的抽象層， 它會引起cpu佔用率增大， 需要關閉。
echo never >/sys/kernel/mm/redhat_transparent_hugepage/defrag
echo never >/sys/kernel/mm/redhat_transparent_hugepage/enabled
echo never >/sys/kernel/mm/transparent_hugepage/enabled
echo never >/sys/kernel/mm/transparent_hugepage/defrag
四、 應用角度進行調優
(1)避免不必要的Reduce任務
如果要處理的資料是排序且已經分割槽的，或者對於一份資料, 需要多次處理, 可以先排序分割槽；然後自定義InputSplit, 將單個分割槽作為單個mapred的輸入；在map中處理資料, Reducer設定為空。這樣, 既重用了已有的 “排序”, 也避免了多餘的reduce任務。
(2)外部檔案引入
有些應用程式要使用外部檔案，如字典，配置檔案等，這些檔案需要在所有task之間共享，可以放到分散式快取DistributedCache中（或直接採用-files選項，機制相同）。更多的這方面的優化方法，還需要在實踐中不斷積累。
(3) 為job新增一個Combiner
為job新增一個combiner可以大大減少shuffle階段從map task拷貝給遠端reduce task的資料量。一般而言，combiner與reducer相同。
(4) 根據處理資料特徵使用最適合和簡潔的Writable型別
Text物件使用起來很方便，但它在由數值轉換到文字或是由UTF8字串轉換到文字時都是低效的，且會消耗大量的CPU時間。當處理那些非文字的資料時，可以使用二進位制的Writable型別，如IntWritable， FloatWritable等。二進位制writable好處：避免檔案轉換的消耗；使map task中間結果佔用更少的空間。
(5)重用Writable型別
很多MapReduce使用者常犯的一個錯誤是，在一個map/reduce方法中為每個輸出都建立Writable物件。
例如，你的Wordcoutmapper方法可能這樣寫：
public voidmap(LongWritable key, Text value, Context context) {
for (String word : words) {
output.collect(new Text(word), newIntWritable(1));
}
}
這樣會導致程式分配出成千上萬個短週期的物件。Java垃圾收集器就要為此做很多的工作。
更有效的寫法是：
class WordCountMapperextends Mapper{
Text wordText = new Text();
IntWritable one = new IntWritable(1);
public void map(LongWritable key, Textvalue, Context context) {
for (String word: words) {
wordText.set(word);
output.collect(wordText,one);
}
}
}
(6)使用StringBuffer而不是String
當需要對字串進行操作時，使用StringBuffer而不是String，String是read-only的，如果對它進行修改，會產生臨時物件，而StringBuffer是可修改的，不會產生臨時物件。
(7) 除錯
最重要，也是最基本的，是要掌握MapReduce程式除錯方法，跟蹤程式的瓶頸。
具體可參考：
http://www.cloudera.com/blog/2009/12/7-tips-for-improving-mapreduce-performance/
五、 Hadoop引數調優
5.1 Linux檔案系統引數調整
(1) noatime 和 nodiratime屬性
檔案掛載時設定這兩個屬性可以明顯提高效能。。預設情況下，Linuxext2/ext3 檔案系統在檔案被訪問、建立、修改時會記錄下檔案的時間戳，比如：檔案建立時間、最近一次修改時間和最近一次訪問時間。如果系統執行時要訪問大量檔案，關閉這些操作，可提升檔案系統的效能。Linux 提供了 noatime 這個引數來禁止記錄最近一次訪問時間戳。
(2) readahead buffer
調整linux檔案系統中預讀緩衝區地大小，可以明顯提高順序讀檔案的效能。預設buffer大小為256 sectors，可以增大為1024或者2408 sectors（注意，並不是越大越好）。可使用blockdev命令進行調整。
(3) 避免RAID和LVM操作
避免在TaskTracker和DataNode的機器上執行RAID和LVM操作，這通常會降低效能。
5.2 Hadoop通用引數調整
(1) dfs.namenode.handler.count或mapred.job.tracker.handler.count
namenode或者jobtracker中用於處理RPC的執行緒數，預設是10，較大叢集，可調大些，比如50。
(2) dfs.datanode.handler.count
datanode上用於處理RPC的執行緒數。預設為3，較大叢集，可適當調大些，比如8。需要注意的是，每新增一個執行緒，需要的記憶體增加。
(3) tasktracker.http.threads
HTTP server上的執行緒數。執行在每個TaskTracker上，用於處理map task輸出。大叢集，可以將其設為40~50。
5.3 HDFS相關配置
(1) dfs.replication
檔案副本數，通常設為3，不推薦修改。
(2) dfs.block.size
HDFS中資料block大小，預設為128M，對於較大叢集，可設為256MB或者512MB。（也可以通過引數mapred.min.split.size配置）
(3) mapred.local.dir和dfs.data.dir
這兩個引數mapred.local.dir和dfs.data.dir 配置的值應當是分佈在各個磁碟上目錄，這樣可以充分利用節點的IO讀寫能力。執行 Linux sysstat包下的iostat -dx 5命令可以讓每個磁碟都顯示它的利用率。
5.4 map/reduce 相關配置
(1) {map/reduce}.tasks.maximum
同時執行在TaskTracker上的最大map/reduce task數，一般設為(core_per_node)/2~2*（cores_per_node）。
(2) io.sort.factor
當一個map task執行完之後，本地磁碟上(mapred.local.dir)有若干個spill檔案，map task最後做的一件事就是執行merge sort，把這些spill檔案合成一個檔案（partition）。執行merge sort的時候，每次同時開啟多少個spill檔案由該引數決定。開啟的檔案越多，不一定merge sort就越快，所以要根據資料情況適當的調整。
(3) mapred.child.java.opts
設定JVM堆的最大可用記憶體，需從應用程式角度進行配置。
5..5 map task相關配置
(1) io.sort.mb
Map task的輸出結果和元資料在記憶體中所佔的buffer總大小。預設為100M，對於大叢集，可設為200M。當buffer達到一定閾值，會啟動一個後臺執行緒來對buffer的內容進行排序，然後寫入本地磁碟(一個spill檔案)。
(2) io.sort.spill.percent
這個值就是上述buffer的閾值，預設是0.8，即80%，當buffer中的資料達到這個閾值，後臺執行緒會起來對buffer中已有的資料進行排序，然後寫入磁碟。
(3) io.sort.record
Io.sort.mb中分配給元資料的記憶體百分比，預設是0.05。這個需要根據應用程式進行調整。
(4)mapred.compress.map.output/Mapred.output.compress
中間結果和最終結果是否要進行壓縮，如果是，指定壓縮方式（Mapred.compress.map.output.codec/Mapred.output.compress.codec）。推薦使用LZO壓縮。Intel內部測試表明，相比未壓縮，使用LZO壓縮的TeraSort作業執行時間減少60%，且明顯快於Zlib壓縮。
5.6 reduce task相關配置
(1) Mapred.reduce.parallel
Reduce shuffle階段copier執行緒數。預設是5，對於較大叢集，可調整為16~25。
5.7 YARN調優
Yarn的資源表示模型為ceontainer（容器）,container 將資源抽象為兩個維度，記憶體和虛擬cpu(vcore)
1. 相容各種計算框架
2. 動態分配資源，減少資源浪費
容器記憶體
yarn.nodemanager.resource.memory-mb
最小容器記憶體
yarn.scheduler.minimum-allocation-mb
容器記憶體增量
yarn.scheduler.increment-allocation-mb
最大容器記憶體
yarn.scheduler.maximum-allocation-mb
容器虛擬cpu核心
yarn.nodemanager.resource.cpu-vcores
最小容器虛擬cpu核心數量
yarn.scheduler.minimum-allocation-vcores
容器虛擬cpu核心增量
yarn.scheduler.increment-allocation-vcores
最大容器虛擬cpu核心數量
yarn.scheduler.maximum-allocation-vcores
MapReduce調優，調優三大原則
1.增大作業並行程度
2.給每個任務足夠的資源
3. 在滿足前2個條件下，儘可能的給shuffle預留資源
六、 從系統實現角度進行優化
6.1 在可移植性和效能之間進行權衡
(1) 排程延遲
Hadoop採用的是動態排程演算法，即：當某個tasktracker上出現空slot時，它會通過HEARBEAT（預設時間間隔為3s，當叢集變大時，會適當調大）告訴jobtracker，之後jobtracker採用某種排程策略從待選task中選擇一個，再通過HEARBEAT告訴tasktracker。從整個過程看，HDFS在獲取下一個task之前，一直處於等待狀態，這造成了資源利用率不高。此外，由於tasktracker獲取新task後，其資料讀取過程是完全序列化的，即：tasktracker獲取task後，依次連線namenode，連線datanode並讀取資料，處理資料。在此過程中，當tasktracker連線namenode和datanode時，HDFS仍在處於等待狀態。
為了解決排程延遲問題，可以考慮的解決方案有：重疊I/O和CPU階段（pipelining），task預取（task prefetching），資料預取（data prefetching）等
(2) 可移植性假設
為了增加Hadoop的可移植性，它採用java語言編寫，這實際上也潛在的造成了HDFS低效。Java儘管可以讓Hadoop的可移植性增強，但是它遮蔽了底層檔案系統，這使它沒法利用一些底層的API對資料儲存和讀寫進行優化。首先，在共享叢集環境下，大量併發讀寫會增加隨機尋道，這大大降低讀寫效率；另外，併發寫會增加磁碟碎片，這將增加讀取代價（HDFS適合檔案順序讀取）。
為了解決該問題，可以考慮的解決方案有：修改tasktracker上的執行緒模型，現在Hadoop上的採用的模型是one thread per client，即每個client連線由一個執行緒處理（包括接受請求，處理請求，返回結果）；修改之後，可將執行緒分成兩組，一組用於處理client通訊（Client Thread），一組用於存取資料（Disk Threads，可採用one thread per disk）。
6.2 Prefetching與preshuffling
(1) PreFetching
preFetching包括Block-intra prefetching和Block-interprefetching:
Block-intraPrefetching對block內部資料處理方式進行優化。採用的策略是以雙向處理（bi-directional processing）方式提升效率，即一端進行計算，一端預取將要用到的資料（同步機制）。
需解決兩個問題，一是計算和預取同步。借用進度條（processing bar）的概念，進度條監控兩端的進度，當同步將被打破時，呼叫一個訊號。二是確定合適的預取率。通過實驗發現，預取資料量並不是越多越好。採用重複實驗的方法確定預取資料率。
Block-interPrefetching在block層面預取資料。當某個task正在處理資料塊A1時，預測器預測它接下來要處理的資料塊，假設是A2，A3，A4,則將這幾個資料塊讀到task所在的rack上，這樣加快了task接下來資料讀取速度。
(2) PreShuffling
資料被map task處理之前，由預測器判斷每條記錄將要被哪個reduce task處理，將這些資料交由靠近該reduce task的節點上的map task處理。
主頁：http://incubator.apache.org/projects/hama.html
6.3 Five Factors
影響Hadoop效能的因素，分別為計算模型，I/O模型，資料解析，索引和排程，同時針對這5個因素提高了相應的提高效能的方法，最後實驗證明，通過這些方法可以將Hadoop效能提高2.5到3.5倍。
(1)計算模型
在Hadoop中，map task產生的中間結果經過sort-merge策略處理後交給reduce task。而這種處理策略（指sort-merge）不能夠定製，這對於有些應用而言（有些應用程式可能不需要排序處理），效能不佳。此外，即使是需要排序歸併處理的，sort-merge也並不是最好的策略。
(2)I/O模型
Reader可以採用兩種方式從底層的儲存系統中讀取資料：direct I/O和streaming I/O。direct I/O是指reader直接從本地檔案中讀取資料；streaming I/O指使用某種程序間通訊方式（如TCP或者JDBC）從另外一個程序中獲取資料。從效能角度考慮，direct I/O效能更高，各種資料庫系統都是採用direct I/O模式。但從儲存獨立性考慮，streaming I/O使Hadoop能夠從任何程序獲取資料，如datanode或database，此外，如果reader不得不從遠端節點上讀取資料，streaming I/O是僅有的選擇。
(3)資料解析
在hadoop中，原始資料要被轉換成key/value的形式以便進一步處理，這就是資料解析。現在有兩種資料解析方法：immutabledecoding and mutable decoding。Hadoop是採用java語言編寫的，java中很多物件是immutable，如String。當用戶試圖修改一個String內容時，原始物件會被丟棄而新物件會被建立以儲存新內容。在Hadoop中，採用了immutable物件儲存字串，這樣每解析一個record就會建立一個新的物件，這就導致了效能低下。
(4)索引
HDFS設計初衷是處理無結構化資料，既然這樣，怎麼可能為資料新增索引。實際上，考慮到以下幾個因素，仍可以給資料新增索引：
A、hadoop提供了結構將資料記錄解析成key/value對，這樣也許可以給key新增索引。
B、如果作業的輸入是一系列索引檔案，可以實現一個新的reader高效處理這些檔案。
(5)排程
Hadoop採用的是動態排程策略，即每次排程一個task執行，這樣會帶來部分開銷。而database採用的靜態排程的策略，即在編譯的時候就確定了排程方案。當用戶提交一個sql時，優化器會生成一個分散式查詢計劃交給每一個節點進行處理。
總結
本文件介紹Hadoop現有的優化點，總體來說，對於Hadoop平臺，現在主要有三種優化思路，分別為：從應用程式角度角度進行優化；從引數配置角度進行優化；從系統實現角度進行優化。對於第一種思路，需要根據具體應用需求而定，同時也需要在長期實踐中積累和總結；對於第二種思路，大部分採用的方法是根據自己叢集硬體和具體應用調整引數，找到一個最優的。對於第三種思路，難度較大，但效果往往非常明顯，總結這方面的優化思路，主要有以下幾個：
(1) 對namenode進行優化，包括增加其吞吐率和解決其單點故障問題。當前主要解決方案有3種：分散式namenode，namenode熱備和zookeeper。
（2）HDFS小檔案問題。當Hadoop中儲存大量小檔案時，namenode擴充套件性和效能受到極大制約。現在Hadoop中已有的解決方案包括：Hadoop Archive，Sequence file和CombineFileInputFormat。
（3）排程框架優化。在Hadoop中，每當出現一個空閒slot後，tasktracker都需要通過HEARBEAT向jobtracker所要task，這個過程的延遲比較大。可以用task預排程的策略解決該問題。
（4）共享環境下的檔案併發存取。在共享環境下，HDFS的隨機尋道次數增加，這大大降低了檔案存取效率。可以通過優化磁碟排程策略的方法改進。
（5） 索引。索引可以大大提高資料讀取效率，如果能根據實際應用需求，為HDFS上的資料新增索引，將大大提高效率。