Hadoop之MapReduce性能調(diào)優(yōu)

    基于對這些組件的深入理解，用戶可以很容易通過調(diào)整一些關(guān)鍵參數(shù)使作業(yè)運行效率達到最優(yōu)，本文將分別從Hadoop管理員和用戶角度介紹如何對Hadoop進行性能調(diào)優(yōu)以滿足各自的需求。

1 概述

Hadoop性能調(diào)優(yōu)是一項工程浩大的工作，它不僅涉及Hadoop本身的性能調(diào)優(yōu)，還涉及更加底層的硬件、操作系統(tǒng)和Java虛擬機等系統(tǒng)的調(diào)優(yōu)。對這幾個系統(tǒng)適當?shù)剡M行調(diào)優(yōu)均有可能給Hadoop帶來性能提升。

對于非Hadoop自身方面的性能調(diào)優(yōu)，比如硬件、操作系統(tǒng)（IOScheduler、啟用預(yù)讀取機制、關(guān)閉Swap等）、Java虛擬機。本文將重點介紹如何通過調(diào)整Hadoop自帶的一些參數(shù)使作業(yè)運行效率達到最優(yōu)?？傮w來說，提高作業(yè)運行效率需要Hadoop管理員和作業(yè)擁有者共同的努力。其中，管理員負責為用戶提供一個高效的作業(yè)環(huán)境，而用戶則負責根據(jù)自己作業(yè)的特點讓它盡可能的運行完成。

    在撰寫本文之時，Apache Hadoop主要分為0.20.x、0.21.x、0.22.x和0.23.x四個系列，Cloudera Hadoop主要分為CDH3和CDH4兩個系列，其中0.23.x和CDH4屬于下一代MapReduce。

2 從管理員角度進行調(diào)優(yōu)

2.1 硬件選擇

Hadoop自身架構(gòu)的基本特點決定了其硬件配置的選項。Hadoop采用了master和slave架構(gòu)，其中，master(JobTracker和NameNode)維護了全局元數(shù)據(jù)信息，重要性遠遠大于slave（TaskTracker和DataNode）。在較低Hadoop版本中，master均存在單點故障問題，master的配置應(yīng)遠遠好于各個slave（TaskTracker或者DataNode）。

2.2 操作系統(tǒng)參數(shù)調(diào)優(yōu)

由于Hadoop自身的特點，它只適合于將Linux作為操作系統(tǒng)的生產(chǎn)環(huán)境。在實際應(yīng)用場景中，管理員適當對linux內(nèi)核參數(shù)進行優(yōu)化，可在一定程度上提高作業(yè)的運行效率，比較有用的調(diào)整選項如下。

1、增大同時打開的文件描述符和網(wǎng)絡(luò)連接上限。

在Hadoop集群中，由于涉及的作業(yè)和任務(wù)數(shù)目非常多，對于某個節(jié)點，由于操作系統(tǒng)內(nèi)核在文件描述符和網(wǎng)絡(luò)連接數(shù)目等方面的限制，大量的文件讀寫操作網(wǎng)絡(luò)連接可能導致作業(yè)失敗，因此，管理員在啟動Hadoop集群時，可以使用ulimit命令將允許同時打開的文件描述符數(shù)目上限增大至一個合適的值，同時調(diào)整內(nèi)核參數(shù)net.core.somaxconn至一個足夠大的值。

此外，Hadoop RPC采用了epoll高并發(fā)庫，如果你使用的Linux內(nèi)核版本在2.6.28以上，你需要適當調(diào)整epoll的文件描述符上限。

2、關(guān)閉swap分區(qū)

在Linux中，如果一個進程的內(nèi)存空間不足，那么，它會將內(nèi)存中的部分數(shù)據(jù)暫時寫入到磁盤上，當需要時，再將磁盤上的數(shù)據(jù)動態(tài)置換到內(nèi)存中，通常而言，這種行為會大大降低進程的執(zhí)行效率。在MapReduce分布式計算環(huán)境中，用戶完全可以通過控制每個作業(yè)處理的數(shù)據(jù)量和每個任務(wù)運行過程中用到的各種緩沖區(qū)中，避免使用swap分區(qū)。具體方式是調(diào)整/etc/sysctl.conf文件中的vm.swappiness。

3、設(shè)置合理的預(yù)讀取緩沖區(qū)大小

磁盤I/O性能的發(fā)展遠遠滯后于CPU和內(nèi)存，因而成為現(xiàn)代計算機系統(tǒng)的一個主要瓶頸。預(yù)讀取可以有效的減少磁盤的尋道次數(shù)和應(yīng)用程序的I/O等待時間，是改進磁盤讀I/O性能的重要優(yōu)化手段之一。管理員可以使用Linux命令blockdev設(shè)置預(yù)讀取緩沖區(qū)的大小，以提高Hadoop中大文件的順序讀的性能。當然，也可以職位Hadoop系統(tǒng)本身增加預(yù)讀緩沖區(qū)大小。

4、文件系統(tǒng)選擇與配置

Hadoop的I/O性能很大程度上依賴于Linux本地文件系統(tǒng)的讀寫性能。Linux中有很多種文件系統(tǒng)可供選擇，比如ext3和ext4，不同的文件系統(tǒng)性能能有一定的差別。

5、I/O調(diào)度器選擇

管理員可以根據(jù)自己的應(yīng)用特點啟用最合適的I/O調(diào)度器。

2.3 JVM參數(shù)調(diào)優(yōu)

由于Hadoop每個服務(wù)和任務(wù)均會運行在一個的JVM中，因此，JVM的一些重要參數(shù)也會影響Hadoop性能。管理員可以通過調(diào)整通過JVM FLAGS和JVM垃圾回收機制提高Hadoop性能。

2.4 Hadoop參數(shù)調(diào)優(yōu)

 1、合理規(guī)劃資源

A、合理設(shè)置合理的槽位數(shù)目

在Hadoop中，計算資源使用槽位（slot）表示的。Slot分為兩種：Mapslot和Reduce slot。每種slot代表了一定量的資源，并且同種slot是同質(zhì)的，也就是說，同種slot代表的資源量是相同的。管理員需要根據(jù)實際需要為TaskTracker配置一定數(shù)目的Map slot和Reduce slot數(shù)目，從而限制每個TaskTracker上并發(fā)的Map Task和Reduce Task數(shù)目。

槽位數(shù)目是在各個TaskTracker上的mapred-site.xml配置的，具體值為：mapreduce.tasktracker.map.tasks.maxium、mapreduce.tasktracker.reduce.tasks.maxium.

B、編寫健康監(jiān)測腳本

Hadoop允許管理員為每個TaskTracker配置一個節(jié)點健康狀況監(jiān)測腳本。TaskTracker中包含一個專門的線程周期性執(zhí)行該腳本，并將腳本執(zhí)行結(jié)果通過心跳機制匯報給JobTracker。一旦JobTracker發(fā)現(xiàn)某個TaskTracker的當前狀況為不健康（比如內(nèi)存或者CPU使用率過高），則會將其加入黑名單，從此不再為它分配新的任務(wù)（當前正在執(zhí)行的任務(wù)仍會正常執(zhí)行完畢），直到該腳本執(zhí)行結(jié)果顯示為健康。

需要注意的是，該機制只有Hadoop0.20.2以上版本中有。

2、調(diào)整心跳配置

A、調(diào)整心跳間隔

TaskTracker與JobTracker之間的心跳間隔大小應(yīng)該適度。

如果太小，JobTracker需要處理高并發(fā)的心跳信息，勢必造成不小的壓力。

如果太大，則空閑的資源不能及時通知JobTracker（進而為之分配新的Task），造成資源空閑，進而降低系統(tǒng)吞吐率。

對于中小規(guī)模（300各節(jié)點以下）的Hadoop集群，縮短TaskTracker與JobTracker之間的心跳間隔可以明顯提高系統(tǒng)吞吐率。

在Hadoop1.0以及更低的版本中，當節(jié)點集群規(guī)模小于300個節(jié)點時，心跳間隔將一直是3秒（不能修改）。這意味著，如果你的集群有10個節(jié)點，那么JobTracker平均每秒也只需要處理3.3（10/3）個心跳請求。如果你的集群有100個節(jié)點，那么JobTracker平均每秒也只需要處理33（100/3）個心跳請求。

對于一臺普通的服務(wù)器，這樣的負載過低，完全沒有充分利用服務(wù)器資源。綜上所述，對于中小規(guī)模的Hadoop集群，3秒的心跳間隔過大，管理員可根據(jù)需要適當減少心跳間隔。

具體配置：mapreduce.jobtracker.heartbeat.interval.min、

mapreduce.heartbeats.in.second、

mapreduce.jobtracker.heartbeats.scaling.factor.

B、啟用帶外心跳

通常而言，心跳是由各個TaskTracker以固定時間間隔為周期發(fā)送給JobTracker的，心跳包含節(jié)點資源使用情況、各任務(wù)運行狀態(tài)等信息。心跳機制是典型的pull-based模型。TaskTracker周期性通過心跳向JobTracker匯報信息，同時獲取新分配的任務(wù)。這種模型使得任務(wù)分配過程存在較大延時：當TaskTracker出現(xiàn)空閑資源，它只能通過下一次心跳。

帶外心跳不同于常規(guī)心跳，它是任務(wù)結(jié)束或者任務(wù)運行失敗時觸發(fā)的，能夠在空閑資源時第一時間通知JobTracker，以便它能夠迅速為空閑資源分配新的任務(wù)。

具體配置：mapreduce.tasktracker.outofband.heartbeat=true

3、磁盤塊配置

Map Task中間結(jié)果要寫到本地磁盤上，對于I/O密集型的任務(wù)來說，這部分數(shù)據(jù)會對本地磁盤造成很大壓力，管理員可通過配置多塊磁盤緩解寫壓力。當存在多塊可用磁盤時，Hadoop將采用輪詢的方式將不同Map Task的中間結(jié)果寫到這些磁盤上，從而平攤負載。

4、設(shè)置合理的RPC Handler和HTTP線程數(shù)目

A、配置RPC Handler數(shù)目

JobTracker需要并發(fā)處理來自各個TaskTracker的RPC請求，管理員可根據(jù)集群規(guī)模和服務(wù)器并發(fā)處理能夠調(diào)整RPC Handler數(shù)目，以使JobTracker服務(wù)能力最佳。

具體配置：

mapred.job.tracker.handler.count;

mapreduce.jobtracker.handler.count

B、配置HTTP線程數(shù)目

在Shuffle階段，Reduce Task通過HTTP請求從各個TaskTracker上讀取Map Task中間結(jié)果，而每個TaskTracker通過TaskTracker通過Jetty Server處理這些HTTP請求。管理員可適當調(diào)整Jetty Server的工作線程數(shù)以提高Jetty Server的并發(fā)處理能力。

具體配置：tasktracker.http.threads

Mapreduce.tasktracker.http.threads.

5、慎用黑名單機制

當一個作業(yè)運行結(jié)束時，它會統(tǒng)計在各個TaskTracker上失敗的任務(wù)數(shù)目。

如果一個TaskTracker失敗的任務(wù)數(shù)目超過一定值，則作業(yè)會將它加到自己的黑名單中。

如果一個TaskTracker被一定數(shù)目的作業(yè)加入黑名單，則JobTracker會將該TaskTracker加入系統(tǒng)黑名單，此后JobTracker不再為其分配新的任務(wù)，直到一定時間段內(nèi)沒出現(xiàn)失敗任務(wù)。

當Hadoop集群規(guī)模較小時，如果一定數(shù)量的節(jié)點被頻繁加入系統(tǒng)黑名單中，則會大大降低集群吞吐率和計算能力，因此建議關(guān)閉該功能。

6、啟用批量任務(wù)調(diào)度

在Hadoop中，調(diào)度器是最核心的組件之一，它負責將系統(tǒng)中空閑的資源分配給各個任務(wù)。當前Hadoop提供了多種調(diào)度器，包括默認的FIFO調(diào)度器、Fair Scheduler、Capacity Scheduler等，調(diào)度器的調(diào)度效率直接決定了系統(tǒng)的吞吐率高低。通常而言，為了將空閑資源盡可能分配任務(wù)，Hadoop調(diào)度器均支持批量任務(wù)調(diào)度。即一次將所有空閑任務(wù)分配下去，而不是一次只分配一個。

7、選擇合適的壓縮算法

Hadoop通常用于處理I/ O密集型應(yīng)用。對于這樣的應(yīng)用，Map Task會輸出大量中間數(shù)據(jù)，這些數(shù)據(jù)的讀寫對用戶是透明，如果能夠支持中間數(shù)據(jù)壓縮存儲，則會明顯提升系統(tǒng)的I/O性能。

當選擇壓縮算法時，需要考慮壓縮比和壓縮效率兩個因素。

有的壓縮算法有很好的壓縮比，但是解壓縮率很低。反之，有一些算法的壓縮/解壓縮效率很高，到那時壓縮比很低。因此，一個優(yōu)秀的壓縮算法需要平衡壓縮比和壓縮效率兩個因素。

具體配置：

mapred.compress.map.out 表示是否要壓縮MapTask中間輸出結(jié)果

mapred.map.output.compression.codec 表示采用的編碼/×××

8、啟用預(yù)讀取機制

前面提到，預(yù)讀取機制可以有效提高磁盤的I/O讀性能。由于Hadoop是典型的順序讀系統(tǒng)，采用預(yù)讀取機制可明顯提高HDFS讀性能和MapReduce作業(yè)執(zhí)行效率，管理員可為MapReduce的數(shù)據(jù)拷貝和IFile文件讀取啟用預(yù)讀取功能。

具體表

3 從用戶角度進行調(diào)優(yōu)

3.1 應(yīng)用程序編寫規(guī)范

從用戶角度來看，除作業(yè)配置參數(shù)以外，應(yīng)用程序本身的編寫方式對性能影響也是非常大的，在編寫應(yīng)用程序的過程中，謹記以下幾條規(guī)則對提高作業(yè)性能是十分有幫助的。

1、設(shè)置Combiner

對于一大批MapReduce應(yīng)用程序，如果可以設(shè)置一個Combiner，那么對于提高作業(yè)性能十分有幫助。Combiner可減少Map Task中間輸出結(jié)果，從而減少各個Reduce Task的遠程拷貝數(shù)據(jù)量，最終表現(xiàn)為Map Task和Reduce Task執(zhí)行時間縮短。

2、選擇合理的Writable類型

在MR模型中，Map Task和Reduce Task的輸入和輸出數(shù)據(jù)類型為Writable類型。Hadoop本身已經(jīng)提供了很多Writable實現(xiàn)，包括基礎(chǔ)類型和對象類型。為應(yīng)用程序處理的數(shù)據(jù)類型選擇合適的Writable類型可大大提升性能。

比如，處理整型數(shù)據(jù)時，直接采用IntWritable比先以Text類型讀取再轉(zhuǎn)換成整型要高效。

如果輸出的整型大部分可用一個或者兩個字節(jié)保存，那么可以直接采用VIntWritable或者VlongWritable。它們采用了變長整型編碼方式，可大大減少輸出數(shù)據(jù)量。

3.2 作業(yè)級別參數(shù)調(diào)優(yōu)

1、規(guī)劃合理的任務(wù)數(shù)目

一個作業(yè)的任務(wù)數(shù)目對作業(yè)運行時間有重要的影響。如果一個作業(yè)的任務(wù)數(shù)目過多（這意味著每個任務(wù)處理數(shù)據(jù)很少，執(zhí)行時間很短），則任務(wù)啟動時間所占比例將會大大增加；反之一個作業(yè)的任務(wù)數(shù)目過少（這意味著每個任務(wù)處理數(shù)據(jù)很多，執(zhí)行時間很長），則可能產(chǎn)生過多的溢寫數(shù)據(jù)影響任務(wù)執(zhí)行性能，且任務(wù)失敗后重新計算代價過大。

在Hadoop中，每個Map Task處理一個Input Split。Input Split的劃分方式是由用戶自定義的InputFormat決定的，默認情況下，由以下三個配置參數(shù)決定：

mapred.min.split.size:Input Split的最小值（在mapred-site.xml配置）

mapred.max.split.size:Input Split的最大值（在mapred-site.xml配置）

dfs.block.size：HDFS中一個block大小

對于Reduce Task而言，每個作業(yè)的Reduce Task數(shù)目通常由用戶決定。用戶可以根據(jù)估量的Map Task輸出數(shù)據(jù)量設(shè)置Reduce Task數(shù)目，以防止每個Reduce Task的數(shù)據(jù)量過大造成大量寫磁盤操作。

2、增加文件副本數(shù)

如果一個作業(yè)并執(zhí)行的任務(wù)數(shù)量非常多，那么這些任務(wù)共同的輸入文件可能成為瓶頸。為防止多個任務(wù)并行讀取一個文件內(nèi)容造成瓶頸，用戶可以根據(jù)需要增加輸入文件的副本數(shù)量。

用戶可以在客戶端配置文件hdfs-site.xml中增dfs.replication選項修改文件副本數(shù)。
   3、啟用推測執(zhí)行機制

推測執(zhí)行是Hadoop對拖后腿任務(wù)的一種優(yōu)化機制。當一個作業(yè)的某些任務(wù)運行速度明顯慢于同作業(yè)的其他任務(wù)時，Hadoop會在另一個節(jié)點上為慢任務(wù)啟動一個備份任務(wù)，這樣，兩個任務(wù)同時處理一份數(shù)據(jù)，而Hadoop最終會將優(yōu)先完成的那個任務(wù)的結(jié)果作為最終結(jié)果，并將另外一個任務(wù)殺掉。

4、設(shè)置失敗容忍度

Hadoop允許設(shè)置作業(yè)級別和任務(wù)級別的失敗容忍度。

作業(yè)級別的失敗容忍是指Hadoop允許每個作業(yè)有一定比例的任務(wù)運行失敗，這部分任務(wù)對應(yīng)的輸入數(shù)據(jù)將被忽略；

任務(wù)級別的失敗容忍是指Hadoop允許任務(wù)運行失敗后再次在另外節(jié)點上嘗試運行，如果一個任務(wù)經(jīng)過若干次嘗試運行后仍然運行失敗，那么Hadoop才會最終認為該任務(wù)運行失敗。

5、適當打開JVM重用功能

為了實現(xiàn)任務(wù)隔離，Hadoop將每個任務(wù)放到一個單獨的JVM執(zhí)行，而對于執(zhí)行時間較短的任務(wù)，JVM啟動和關(guān)閉將占用很大比例的時間，為此，用戶可啟用JVM重用功能，這樣，一個JVM可以連續(xù)啟動多個同類型任務(wù)。

6、設(shè)置任務(wù)超時時間

在一些特殊情況下，一個任務(wù)可能因為某種原因阻塞了，這回拖慢整個作業(yè)的執(zhí)行進度，甚至可能導致作業(yè)無法作業(yè)運行結(jié)束。針對這種情況，Hadoop增加了任務(wù)超時機制。如果一個任務(wù)在一定時間間隔內(nèi)沒有匯報進度，則TaskTracker會主動將其殺死，從而在另外一個節(jié)點上重新啟動執(zhí)行。

7、合理使用DistributedCache

當用戶的應(yīng)用程序需要一個外部文件（比如數(shù)據(jù)字典、配置文件等）時，通常需要使用DistributedCache將文件分發(fā)到各個節(jié)點上。一般情況下，得到外部文件有兩種方法：一種是外部文件與應(yīng)用程序jar包一起放到客戶端，當提交作業(yè)由客戶端上傳到HDFS的一個目錄下，然后通過DistributedCache分發(fā)到各個節(jié)點上。另一種方法是事先將外部文件直接放到HDFS上。從效率上講，第二種方法比第一種更高效。第二中方式不僅節(jié)省了客戶端上傳文件的時間，還隱含著告訴DistributedCache，請將文件下載到各節(jié)點的public級別共享目錄上，這樣后續(xù)作業(yè)可重用已經(jīng)下載好的文件，不必重復下載，即一次下載，終身受益。

8、合理控制Reduce Task的啟動時機

在MR計算模型中，由于Reduce Task依賴于Map Task的執(zhí)行結(jié)果，因此，從運算邏輯上講，Reduce Task應(yīng)晚于Map Task啟動。

在Hadoop中，合理控制Reduce Task啟動時機不僅可以加快作業(yè)運行速度，而且可提高系統(tǒng)資源利用率。如果ReduceTask啟動過早，則可能由于Reduce Task長時間占用Resourceslot造成slot Hoarding現(xiàn)象，從而降低資源利用率；反之，如果Reduce Task啟動過晚，則會導致Reduce Task獲取資源延遲，增加了作業(yè)運行時間。

9、跳過壞記錄

Hadoop是用于處理海量數(shù)據(jù)的，對于大部分數(shù)據(jù)密集型應(yīng)用而言，丟棄一條或者幾條數(shù)據(jù)對最終結(jié)果的影響不大，正因為如此，Hadoop為用戶提供了跳過壞記錄的功能。當一條或者幾條數(shù)據(jù)導致任務(wù)運行失敗時，Hadoop可自動識別并跳過這些壞記錄。

10、提高作業(yè)優(yōu)先級

所有Hadoop作業(yè)調(diào)度器進行任務(wù)調(diào)度時均會考慮作業(yè)優(yōu)先級這一因素。一個作業(yè)的優(yōu)先級越高，它能夠獲取的資源（slot）也越多。需要注意的是，通常而言，在生產(chǎn)環(huán)境中，管理員已經(jīng)按照作業(yè)重要程度對作業(yè)進行了分級，不同重要程度的作業(yè)允許配置的優(yōu)先級不同，用戶不可以擅自進行調(diào)整。

Hadoop提供了5種作業(yè)優(yōu)先級，分別是VERY_HIGH、HIGH、NORMAL、LOW和VERY_LOW。

3.3 任務(wù)級別參數(shù)調(diào)優(yōu)

1、Map Task調(diào)優(yōu)

Map Task的輸出結(jié)果將被暫時存放到一個環(huán)形緩沖區(qū)中，這個緩沖區(qū)的大小由參數(shù)“io.sort.mb”指定，默認是100MB，該緩沖區(qū)主要由兩部分組成：索引和實際數(shù)據(jù)。

默認情況下，索引占整個buffer的比例為io.sort.record.percent，5%，剩下的空間全部存放數(shù)據(jù)，當且僅當滿足以下任意一個條件時，才會觸發(fā)一次flush，生成一個臨時文件：

A、索引空間使用率比例（io.sort.spill.percent）為0.8；

B、數(shù)據(jù)空間使用率比例(io.sort.spill.percent)為0.8；

合理調(diào)整io.sort.spill.percent值，可減少中間文件數(shù)目，提高任務(wù)執(zhí)行效率，舉例說明，如果你的key/value非常小，則可以適當調(diào)整io.sort.spill.percent值，以防止索引空間優(yōu)先達到使用上限觸發(fā)flush。考慮到每條數(shù)據(jù)記錄需要占用索引大小為16B，因此，建議io.sort.spill.percent=16/（16+R）,其中R為平均每條記錄的長度。

2、Reduce Task調(diào)優(yōu)

Reduce Task會啟動多個拷貝線程從每個Map Task上讀取相應(yīng)的中間結(jié)果，具體的線程數(shù)目由《mapred.reduce.parallel.copies》（默認為5）指定。

對于每個待拷貝的文件，如果文件大小小于一定閥值A(chǔ)，則將其放到內(nèi)存中，否則以文件的形式存放到磁盤上，

如果內(nèi)存文件滿足一定條件D，則會將這些數(shù)據(jù)寫入磁盤，而當磁盤上文件數(shù)目達到io.sort.factor(默認是10)，進行一次合并。

閥值A(chǔ)為：heapsize*{mapred.job.shuffle.input.buffer.percent}*0.25

其中，heapsize是通過參數(shù)“mapred.child.java.opts”指定的，默認是200MB；mapred.job.shuffle.input.buffer.percent默認大小是0.7。

條件D為以下兩個條件任意中一個：

內(nèi)存使用率達到mapred.job.shuffle.merge.percent的0.06時；

內(nèi)存中文件數(shù)目超過(mapred.inmem.merge.percent)1000時；