Hadoop之MapReduce性能調優

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

1 概述

Hadoop性能調優是一項工程浩大的工作，它不僅涉及Hadoop本身的性能調優，還涉及更加底層的硬件、操作系統和Java虛擬機等系統的調優。對這幾個系統適當地進行調優均有可能給Hadoop帶來性能提升。

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

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

2 從管理員角度進行調優

2.1 硬件選擇

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

2.2 操作系統參數調優

由于Hadoop自身的特點，它只適合于將Linux作為操作系統的生產環境。在實際應用場景中，管理員適當對linux內核參數進行優化，可在一定程度上提高作業的運行效率，比較有用的調整選項如下。

1、增大同時打開的文件描述符和網絡連接上限。

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

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

2、關閉swap分區

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

3、設置合理的預讀取緩沖區大小

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

4、文件系統選擇與配置

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

5、I/O調度器選擇

管理員可以根據自己的應用特點啟用最合適的I/O調度器。

2.3 JVM參數調優

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

2.4 Hadoop參數調優

 1、合理規劃資源

A、合理設置合理的槽位數目

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

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

B、編寫健康監測腳本

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

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

2、調整心跳配置

A、調整心跳間隔

TaskTracker與JobTracker之間的心跳間隔大小應該適度。

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

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

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

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

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

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

mapreduce.heartbeats.in.second、

mapreduce.jobtracker.heartbeats.scaling.factor.

B、啟用帶外心跳

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

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

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

3、磁盤塊配置

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

4、設置合理的RPC Handler和HTTP線程數目

A、配置RPC Handler數目

JobTracker需要并發處理來自各個TaskTracker的RPC請求，管理員可根據集群規模和服務器并發處理能夠調整RPC Handler數目，以使JobTracker服務能力最佳。

具體配置：

mapred.job.tracker.handler.count;

mapreduce.jobtracker.handler.count

B、配置HTTP線程數目

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

具體配置：tasktracker.http.threads

Mapreduce.tasktracker.http.threads.

5、慎用黑名單機制

當一個作業運行結束時，它會統計在各個TaskTracker上失敗的任務數目。

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

如果一個TaskTracker被一定數目的作業加入黑名單，則JobTracker會將該TaskTracker加入系統黑名單，此后JobTracker不再為其分配新的任務，直到一定時間段內沒出現失敗任務。

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

6、啟用批量任務調度

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

7、選擇合適的壓縮算法

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

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

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

具體配置：

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

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

8、啟用預讀取機制

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

具體表

3 從用戶角度進行調優

3.1 應用程序編寫規范

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

1、設置Combiner

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

2、選擇合理的Writable類型

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

比如，處理整型數據時，直接采用IntWritable比先以Text類型讀取再轉換成整型要高效。

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

3.2 作業級別參數調優

1、規劃合理的任務數目

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

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

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而言，每個作業的Reduce Task數目通常由用戶決定。用戶可以根據估量的Map Task輸出數據量設置Reduce Task數目，以防止每個Reduce Task的數據量過大造成大量寫磁盤操作。

2、增加文件副本數

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

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

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

4、設置失敗容忍度

Hadoop允許設置作業級別和任務級別的失敗容忍度。

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

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

5、適當打開JVM重用功能

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

6、設置任務超時時間

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

7、合理使用DistributedCache

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

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

在MR計算模型中，由于Reduce Task依賴于Map Task的執行結果，因此，從運算邏輯上講，Reduce Task應晚于Map Task啟動。

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

9、跳過壞記錄

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

10、提高作業優先級

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

Hadoop提供了5種作業優先級，分別是VERY_HIGH、HIGH、NORMAL、LOW和VERY_LOW。

3.3 任務級別參數調優

1、Map Task調優

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

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

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

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

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

2、Reduce Task調優

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

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

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

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

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

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

內存使用率達到mapred.job.shuffle.merge.percent的0.06時；

內存中文件數目超過(mapred.inmem.merge.percent)1000時；