Hadoop性能調優、YARN的內存和CPU配置

轉自： https://blog.csdn.net/tototuzuoquan/article/details/80671128

 

轉： 
https://blog.csdn.net/dehu_zhou/article/details/52808752 
https://blog.csdn.net/dxl342/article/details/52840455

Hadoop為用戶作業提供了多種可配置的參數，以允許用戶根據作業特點調整這些參數值使作業運行效率達到最優。

一 應用程序編寫規范

1.設置Combiner

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

2. 選擇合理的Writable類型

在MapReduce模型中，Map Task和Reduce Task的輸入和輸出類型均為Writable。Hadoop本身已經提供了很多Writable實現，包括IntWritable、FloatWritable。為應用程序處理的數據選擇合適的Writable類型可大大提升性能。比如處理整數類型數據時，直接采用IntWritable比先以Text類型讀入在轉換為整數類型要高效。如果輸出整數的大部分可用一個或兩個字節保存，那么直接采用VIntWritable或者VLongWritable，它們采用了變長整型的編碼方式，可以大大減少輸出數據量。

二 作業級別參數調優

1.規划合理的任務數目

在Hadoop中，每個Map Task處理一個Input Split。Input Split的划分方式是由用戶自定義的InputFormat決定的，默認情況下，有以下三個參數決定。 
mapred.min.split.size ：Input Split的最小值 默認值1 
mapred.max.split.szie: Input Split的最大值 
dfs.block.size：HDFS 中一個block大小 默認值64MB 
golsize:它是用戶期望的Input Split數目=totalSize/numSplits ,其中totalSize為文件的總大小；numSplits為用戶設定的Map Task個數，默認情況下是1. 
splitSize = max{minSize,min{goalSize,blockSize}} 如果想讓InputSize尺寸大於block尺寸，直接增大配置參數mpared.min.split.size即可。

2.增加輸入文件的副本數

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

3.啟動推測執行機制

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

4.設置失敗容忍度

Hadoop運行設置任務級別和作業級別的失敗容忍度。作業級別的失敗容忍度是指Hadoop允許每個作業有一定比例的任務運行失敗，這部分任務對應的輸入數據將被忽略； 
任務級別的失敗容忍度是指Hadoop允許任務失敗后再在另外節點上嘗試運行，如果一個任務經過若干次嘗試運行后仍然運行失敗，那么Hadoop才會最終認為該任務運行失敗。 
用戶應該根據應用程序的特點設置合理的失敗容忍度，以盡快讓作業運行完成和避免沒必要的資源浪費。

5.適當打開JVM重用功能

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

6.設置任務超時時間

如果一個任務在一定的時間內未匯報進度，則TaskTracker會主動將其殺死，從而在另一個節點上重新啟動執行。用戶可根據實際需要配置任務超時時間。

7.合理使用DistributedCache

一般情況下，得到外部文件有兩種方法：一種是外部文件與應用程序jar包一起放到客戶端，當提交作業時由客戶端上傳到HDFS的一個目錄下，然后通過Distributed Cache分發到各個節點上；另一種方法是事先將外部文件直接放到HDFS上，從效率上講，第二種方法更高效。第二種方法不僅節省了客戶端上傳文件的時間，還隱含着告訴DistributedCache:”請將文件下載到各個節點的pubic級別共享目錄中”，這樣，后續所有的作業可重用已經下載好的文件，不必重復下載。

8.跳過壞記錄

Hadoop為用戶提供了跳過壞記錄的功能，當一條或幾條壞數據記錄導致任務運行失敗時，Hadoop可自動識別並跳過這些壞記錄。

9.提高作業優先級

所有Hadoop作業調度器進行任務調度時均會考慮作業優先級這一因素。作業的優先級越高，它能夠獲取的資源（slot數目)也越多。Hadoop提供了5種作業優先級，分別為VERY_HIGH、 HIGH、 NORMAL、 LOW、 VERY_LOW。 
注：在生產環境中，管理員已經按照作業重要程度對作業進行了分級，不同重要程度的作業允許配置的優先級不同，用戶可以擅自進行調整。

10.合理控制Reduce Task的啟動時機

如果Reduce Task啟動過早，則可能由於Reduce Task長時間占用Reduce slot資源造成”slot Hoarding”現象，從而降低資源利用率；反之，如果Reduce Task啟動過晚，則會導致Reduce Task獲取資源延遲，增加了作業的運行時間。

三 任務級別參數調優

hadoop任務級別參數調優分兩個方面: Map Task和Reduce Task。

1.Map Task調優

map運行階段分為:Read、Map、Collect、Spill、Merge五個階段。
map 任務執行會產生中間數據,但這些中間結果並沒有直接IO到磁盤上,而是先存儲在緩存(buffer)中,並在緩存中進行一些預排序來優化整個map的性能,存儲map中間數據的緩存默認大小為100M，由io.sort.mb 參數指定。這個大小可以根據需要調整。當map任務產生了非常大的中間數據時可以適當調大該參數,使緩存能容納更多的map中間數據，而不至於大頻率的IO磁盤,當系統性能的瓶頸在磁盤IO的速度上,可以適當的調大此參數來減少頻繁的IO帶來的性能障礙。
由於map任務運行時中間結果首先存儲在緩存中,默認當緩存的使用量達到80%(或0.8)的時候就開始寫入磁盤,這個過程叫做spill(也叫溢出),進行spill的緩存大小可以通過io.sort.spill.percent 參數調整，這個參數可以影響spill的頻率。進而可以影響IO的頻率。
當map任務計算成功完成之后，如果map任務有輸出，則會產生多個spill。接下來map必須將些spill進行合並,這個過程叫做merge, merge過程是並行處理spill的,每次並行多少個spill是由參數io.sort.factor指定的默認為10個。但是當spill的數量非常大的時候，merge一次並行運行的spill仍然為10個,這樣仍然會頻繁的IO處理,因此適當的調大每次並行處理的spill數有利於減少merge數因此可以影響map的性能。
當map輸出中間結果的時候也可以配置壓縮。

2. Reduce Task調優

reduce 運行階段分為shuflle(copy) merge sort   reduce write五個階段。
shuffle 階段為reduce 全面拷貝map任務成功結束之后產生的中間結果,如果上面map任務采用了壓縮的方式,那么reduce 將map任務中間結果拷貝過來后首先進行解壓縮,這一切是在reduce的緩存中做的,當然也會占用一部分cpu。為了優化reduce的執行時間,reduce也不是等到所有的map數據都拷貝過來的時候才開始運行reduce任務，而是當job執行完第一個map任務時開始運行的。reduce 在shuffle階段 實際上是從不同的並且已經完成的map上去下載屬於自己的數據,由於map任務數很多,所有這個copy過程是並行的,既同時有許多個reduce取拷貝map，這個並行的線程是通過mapred.reduce.parallel.copies 參數指定，默認為5個,也就是說無論map的任務數是多少個，默認情況下一次只能有5個reduce的線程去拷貝map任務的執行結果。所以當map任務數很多的情況下可以適當的調整該參數，這樣可以讓reduce快速的獲得運行數據來完成任務。
reduce線程在下載map數據的時候也可能因為各種各樣的原因(網絡原因、系統原因等），存儲該map數據所在的datannode 發生了故障，這種情況下reduce任務將得不到該datanode上的數據了,同時該 download thread 會嘗試從別的datanode下載,可以通過mapred.reduce.copy.backoff (默認為30秒)來調整下載線程的下載時間，如果網絡不好的集群可以通過增加該參數的值來增加下載時間,以免因為下載時間過長reduce將該線程判斷為下載失敗。
reduce 下載線程在map結果下載到本地時,由於是多線程並行下載，所以也需要對下載回來的數據進行merge,所以map階段設置的io.sort.factor 也同樣會影響這個reduce的。
同map一樣 該緩沖區大小也不是等到完全被占滿的時候才寫入磁盤而是默認當完成0.66的時候就開始寫磁盤操作,該參數是通過mapred.job.shuffle.merge.percent 指定的。
當reduce 開始進行計算的時候通過mapred.job.reduce.input.buffer.percent 來指定需要多少的內存百分比來作為reduce讀已經sort好的數據的buffer百分比,該值默認為0。Hadoop假設用戶的reduce()函數需要所有的JVM內存，因此執行reduce()函數前要釋放所有內存。如果設置了該值，可將部分文件保存在內存中(不必寫到磁盤上)。

總之，Map Task和Reduce Task調優的一個原則就是減少數據的傳輸量、盡量使用內存、減少磁盤IO的次數、增大任務並行數，除此之外還有根據自己集群及網絡的實際情況來調優。

三 管理員角度調優

管理員負責為用戶作業提供一個高效的運行環境。管理員需要從全局出發，通過調整一些關鍵參數提高系統的吞吐率和性能。總體上來看，管理員需從硬件選擇、操作系統參數調優、JVM參數調優和Hadoop參數調優等四個角度入手，為Hadoop用戶提供一個高效的作業運行環境。

硬件選擇 
Hadoop自身架構的基本特點決定了其硬件配置的選項。Hadoop采用了Master/Slave架構，其中，master維護了全局元數據信息，重要性遠遠大於slave。在較低Hadoop版本中，master存在單點故障問題，因此，master的配置應遠遠好於各個slave。

操作系統參數調優

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

使用ulimit命令將允許同時打開的文件描述符數目上限增大至一個合適的值。同時調整內核參數net.core.somaxconn網絡連接數目至一個足夠大的值。

補充：net.core.somaxconn的作用 
net.core.somaxconn是Linux中的一個kernel參數，表示socket監聽（listen）的backlog上限。什么是backlog呢？backlog就是socket的監聽隊列，當一個請求（request）尚未被處理或建立時，它會進入backlog。而socket server可以一次性處理backlog中的所有請求，處理后的請求不再位於監聽隊列中。當server處理請求較慢，以至於監聽隊列被填滿后，新來的請求會被拒絕。在Hadoop 1.0中，參數ipc.server.listen.queue.size控制了服務端socket的監聽隊列長度，即backlog長度，默認值是128。而Linux的參數net.core.somaxconn默認值同樣為128。當服務端繁忙時，如NameNode或JobTracker，128是遠遠不夠的。這樣就需要增大backlog，例如我們的3000台集群就將ipc.server.listen.queue.size設成了32768，為了使得整個參數達到預期效果，同樣需要將kernel參數net.core.somaxconn設成一個大於等於32768的值。

2.關閉swap分區

避免使用swap分區，提供程序的執行效率。
除此之外，設置合理的預讀取緩沖區的大小、文件系統選擇與配置及I/O調度器選擇等

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

Hadoop參數調優

1.合理規划資源

設置合理的槽位數目
在Hadoop中，計算資源是用槽位表示的。slot分為兩種：Map  Slot和Reduce Slot。每種slot代表一定量的資源，且同種slot是同質的，也就是說，同種slot代表的資源量是相同的。管理員需要根據實際需要為TaskTracker配置一定數目的Map Slot和Reduce Slot數目，從而限制每個TaskTracker上並發執行的Map Task和Reduce Task的數目。
編寫健康監測腳本
Hadoop允許管理員為每個TaskTracker配置一個節點健康狀況監測腳本。TaskTracker中包含一個專門的線程周期性執行該腳本，並將腳本執行結果通過心跳機制匯報給JobTracker。一旦JobTracker發現某個TaskTracker的當前狀況為“不健康”，則會將其加入黑名單，從此不再為它分配任務。

2. 調整心跳配置

調整心跳的間隔 因根據自己集群的規模適度的調整心跳間隔
啟用帶外心跳   為了減少任務分配延遲，Hadoop引入了帶外心跳。帶外心跳不同於常規心跳，它是任務運行結束或者任務運行失敗時觸發的，能夠在出現空閑資源時第一時間通知JobTracker,以便它能夠迅速為空閑資源分配新的任務。

除此之外，還包括磁盤塊配置、設置合理的RPC Handler和HTTP線程數目、慎用黑名單機制、啟用批量任務調度、選擇合適的壓縮算法、啟用預讀取機制等。
注：當一個集群的規模較小時，如果一定數量的節點被頻繁的加入系統黑名單中，則會大大降低集群的吞吐率和計算能力。

四：YARN的內存和CPU配置

Hadoop YARN同時支持內存和CPU兩種資源的調度，本文介紹如何配置YARN對內存和CPU的使用。

YARN作為一個資源調度器，應該考慮到集群里面每一台機子的計算資源，然后根據application申請的資源進行分配Container。Container是YARN里面資源分配的基本單位，具有一定的內存以及CPU資源。

在YARN集群中，平衡內存、CPU、磁盤的資源的很重要的，根據經驗，每兩個container使用一塊磁盤以及一個CPU核的時候可以使集群的資源得到一個比較好的利用。

1、內存配置

關於 內存 相關的配置可以參考hortonwork公司的文檔 Determine HDP Memory Configuration Settings （https://docs.hortonworks.com/HDPDocuments/HDP2/HDP-2.1.1/bk_installing_manually_book/content/rpm-chap1-11.html）來配置你的集群。

YARN以及MAPREDUCE所有可用的內存資源應該要除去系統運行需要的以及其他的hadoop的一些程序，總共保留的內存=系統內存+HBASE內存。

可以參考下面的表格確定應該保留的內存： 

計算每台機子最多可以擁有多少個container，可以使用下面的公式: 
containers = min (2*CORES, 1.8*DISKS, (Total available RAM) / MIN_CONTAINER_SIZE) 
說明：

CORES 為機器CPU核數
DISKS 為機器上掛載的磁盤個數
Total available RAM 為機器總內存 MIN_CONTAINER_SIZE 是指container最小的容量大小，這需要根據具體情況去設置，可以參考下面的表格

每個container的平均使用內存大小計算方式為：

RAM-per-container = max(MIN_CONTAINER_SIZE, (Total Available RAM) / containers))

通過上面的計算，YARN以及MAPREDUCE可以這樣配置： 

舉個例子：對於128G內存、32核CPU的機器，掛載了7個磁盤，根據上面的說明，系統保留內存為24G，不適應HBase情況下，系統剩余可用內存為104G，計算containers值如下：

containers = min (2*32, 1.8* 7 , (128-24)/2) = min (64, 12.6 , 51) = 13

計算RAM-per-container值如下：

RAM-per-container = max (2, (124-24)/13) = max (2, 8) = 8

這樣集群中下面的參數配置值如下： 

你也可以使用腳本 yarn-utils.py （https://docs.hortonworks.com/HDPDocuments/HDP2/HDP-2.1.1/bk_installing_manually_book/content/rpm-chap1-9.html）來計算上面的值：

python yarn-utils.py -c 32 -m 128 -d 7 -k False

返回結果如下：

Using cores=32 memory=128GB disks=7 hbase=False Profile: cores=32 memory=106496MB reserved=24GB usableMem=104GB disks=7 Num Container=13 Container Ram=8192MB Used Ram=104GB Unused Ram=24GB yarn.scheduler.minimum-allocation-mb=8192 yarn.scheduler.maximum-allocation-mb=106496 yarn.nodemanager.resource.memory-mb=106496 mapreduce.map.memory.mb=8192 mapreduce.map.java.opts=-Xmx6553m mapreduce.reduce.memory.mb=8192 mapreduce.reduce.java.opts=-Xmx6553m yarn.app.mapreduce.am.resource.mb=8192 yarn.app.mapreduce.am.command-opts=-Xmx6553m mapreduce.task.io.sort.mb=3276

對應的xml配置為：

<property> <name>yarn.nodemanager.resource.memory-mb</name> <value>106496</value> </property> <property> <name>yarn.scheduler.minimum-allocation-mb</name> <value>8192</value> </property> <property> <name>yarn.scheduler.maximum-allocation-mb</name> <value>106496</value> </property> <property> <name>yarn.app.mapreduce.am.resource.mb</name> <value>8192</value> </property> <property> <name>yarn.app.mapreduce.am.command-opts</name> <value>-Xmx6553m</value> </property>

另外，還有一下幾個參數：

yarn.nodemanager.vmem-pmem-ratio ：任務每使用1MB物理內存，最多可使用虛擬內存量，默認是2.1。 yarn.nodemanager.pmem-check-enabled ：是否啟動一個線程檢查每個任務正使用的物理內存量，如果任務超出分配值，則直接將其殺掉，默認是true。 yarn.nodemanager.vmem-pmem-ratio ：是否啟動一個線程檢查每個任務正使用的虛擬內存量，如果任務超出分配值，則直接將其殺掉，默認是true。

第一個參數的意思是當一個map任務總共分配的物理內存為8G的時候，該任務的container最多內分配的堆內存為6.4G，可以分配的虛擬內存上限為8*2.1=16.8G。另外，照這樣算下去，每個節點上YARN可以啟動的Map數為104/8=13個，似乎偏少了，這主要是和我們掛載的磁盤數太少了有關，人為的調整 RAM-per-container 的值為4G或者更小的一個值是否更合理呢？當然，這個要監控集群實際運行情況來決定了。 
CPU配置 
YARN中目前的CPU被划分成虛擬CPU（CPU virtual Core），這里的虛擬CPU是YARN自己引入的概念，初衷是，考慮到不同節點的CPU性能可能不同，每個CPU具有的計算能力也是不一樣的，比如某個物理CPU的計算能力可能是另外一個物理CPU的2倍，這時候，你可以通過為第一個物理CPU多配置幾個虛擬CPU彌補這種差異。用戶提交作業時，可以指定每個任務需要的虛擬CPU個數。

在YARN中，CPU相關配置參數如下：

yarn.nodemanager.resource.cpu-vcores ：表示該節點上YARN可使用的虛擬CPU個數，默認是8，注意，目前推薦將該值設值為與物理CPU核數數目相同。如果你的節點CPU核數不夠8個，則需要調減小這個值，而YARN不會智能的探測節點的物理CPU總數。 yarn.scheduler.minimum-allocation-vcores ：單個任務可申請的最小虛擬CPU個數，默認是1，如果一個任務申請的CPU個數少於該數，則該對應的值改為這個數。 yarn.scheduler.maximum-allocation-vcores ：單個任務可申請的最多虛擬CPU個數，默認是32。

對於一個CPU核數較多的集群來說，上面的默認配置顯然是不合適的，在我的測試集群中，4個節點每個機器CPU核數為32，可以配置為：

<property> <name>yarn.nodemanager.resource.cpu-vcores</name> <value>32</value> </property> <property> <name>yarn.scheduler.maximum-allocation-vcores</name> <value>128</value> </property>

總結 
根據上面的說明，我的測試集群中集群節點指標如下： 
每個節點分配的物理內存、虛擬內存和CPU核數如下： 
實際生產環境中，可能不會像上面那樣設置，比如不會把所有節點的CPU核數都分配給Spark，需要保留一個核留給系統使用；另外，內存上限也會做些設置。

小結

Hadoop 性能調優是一項工程浩大的工作，它不僅涉及Hadoop本身的性能調優，還涉及更底層的硬件、操作系統和Java虛擬機等系統的調優。
總體來說，提高作業運行效率需要Hadoop管理員和作業擁有者共同的努力，其中，管理員負責為用戶提供一個高效的作業運行環境，而用戶則根