hadoop作業(yè)調(diào)優(yōu)參數(shù)整理及原理是什么

hadoop作業(yè)調(diào)優(yōu)參數(shù)整理及原理是什么，針對(duì)這個(gè)問題，這篇文章詳細(xì)介紹了相對(duì)應(yīng)的分析和解答，希望可以幫助更多想解決這個(gè)問題的小伙伴找到更簡(jiǎn)單易行的方法。

1 Map side tuning參數(shù)

1.1 MapTask運(yùn)行內(nèi)部原理

當(dāng)map task開始運(yùn)算，并產(chǎn)生中間數(shù)據(jù)時(shí)，其產(chǎn)生的中間結(jié)果并非直接就簡(jiǎn)單的寫入磁盤。這中間的過程比較復(fù)雜，并且利用到了內(nèi)存buffer來進(jìn)行已經(jīng)產(chǎn)生的部分結(jié)果的緩存，并在內(nèi)存buffer中進(jìn)行一些預(yù)排序來優(yōu)化整個(gè)map的性能。如上圖所示，每一個(gè)map都會(huì)對(duì)應(yīng)存在一個(gè)內(nèi)存buffer（MapOutputBuffer，即上圖的buffer
in memory），map會(huì)將已經(jīng)產(chǎn)生的部分結(jié)果先寫入到該buffer中，這個(gè)buffer默認(rèn)是100MB大小，但是這個(gè)大小是可以根據(jù)job提交時(shí)的參數(shù)設(shè)定來調(diào)整的，該參數(shù)即為：io.sort.mb。當(dāng)map的產(chǎn)生數(shù)據(jù)非常大時(shí)，并且把io.sort.mb調(diào)大，那么map在整個(gè)計(jì)算過程中spill的次數(shù)就勢(shì)必會(huì)降低，map task對(duì)磁盤的操作就會(huì)變少，如果map tasks的瓶頸在磁盤上，這樣調(diào)整就會(huì)大大提高map的計(jì)算性能。map做sort和spill的內(nèi)存結(jié)構(gòu)如下如所示：

map在運(yùn)行過程中，不停的向該buffer中寫入已有的計(jì)算結(jié)果，但是該buffer并不一定能將全部的map輸出緩存下來，當(dāng)map輸出超出一定閾值（比如100M），那么map就必須將該buffer中的數(shù)據(jù)寫入到磁盤中去，這個(gè)過程在mapreduce中叫做spill。map并不是要等到將該buffer全部寫滿時(shí)才進(jìn)行spill，因?yàn)槿绻繉憹M了再去寫spill，勢(shì)必會(huì)造成map的計(jì)算部分等待buffer釋放空間的情況。所以，map其實(shí)是當(dāng)buffer被寫滿到一定程度（比如80%）時(shí)，就開始進(jìn)行spill。這個(gè)閾值也是由一個(gè)job的配置參數(shù)來控制，即io.sort.spill.percent，默認(rèn)為0.80或80%。這個(gè)參數(shù)同樣也是影響spill頻繁程度，進(jìn)而影響map
task運(yùn)行周期對(duì)磁盤的讀寫頻率的。但非特殊情況下，通常不需要人為的調(diào)整。調(diào)整io.sort.mb對(duì)用戶來說更加方便。

當(dāng)map task的計(jì)算部分全部完成后，如果map有輸出，就會(huì)生成一個(gè)或者多個(gè)spill文件，這些文件就是map的輸出結(jié)果。map在正常退出之前，需要將這些spill合并（merge）成一個(gè)，所以map在結(jié)束之前還有一個(gè)merge的過程。merge的過程中，有一個(gè)參數(shù)可以調(diào)整這個(gè)過程的行為，該參數(shù)為：io.sort.factor。該參數(shù)默認(rèn)為10。它表示當(dāng)merge spill文件時(shí)，最多能有多少并行的stream向merge文件中寫入。比如如果map產(chǎn)生的數(shù)據(jù)非常的大，產(chǎn)生的spill文件大于10，而io.sort.factor使用的是默認(rèn)的10，那么當(dāng)map計(jì)算完成做merge時(shí)，就沒有辦法一次將所有的spill文件merge成一個(gè)，而是會(huì)分多次，每次最多10個(gè)stream。這也就是說，當(dāng)map的中間結(jié)果非常大，調(diào)大io.sort.factor，有利于減少merge次數(shù)，進(jìn)而減少map對(duì)磁盤的讀寫頻率，有可能達(dá)到優(yōu)化作業(yè)的目的。

當(dāng)job指定了combiner的時(shí)候，我們都知道m(xù)ap介紹后會(huì)在map端根據(jù)combiner定義的函數(shù)將map結(jié)果進(jìn)行合并。運(yùn)行combiner函數(shù)的時(shí)機(jī)有可能會(huì)是merge完成之前，或者之后，這個(gè)時(shí)機(jī)可以由一個(gè)參數(shù)控制，即min.num.spill.for.combine（default 3），當(dāng)job中設(shè)定了combiner，并且spill數(shù)最少有3個(gè)的時(shí)候，那么combiner函數(shù)就會(huì)在merge產(chǎn)生結(jié)果文件之前運(yùn)行。通過這樣的方式，就可以在spill非常多需要merge，并且很多數(shù)據(jù)需要做conbine的時(shí)候，減少寫入到磁盤文件的數(shù)據(jù)數(shù)量，同樣是為了減少對(duì)磁盤的讀寫頻率，有可能達(dá)到優(yōu)化作業(yè)的目的。

減少中間結(jié)果讀寫進(jìn)出磁盤的方法不止這些，還有就是壓縮。也就是說map的中間，無論是spill的時(shí)候，還是最后merge產(chǎn)生的結(jié)果文件，都是可以壓縮的。壓縮的好處在于，通過壓縮減少寫入讀出磁盤的數(shù)據(jù)量。對(duì)中間結(jié)果非常大，磁盤速度成為map執(zhí)行瓶頸的job，尤其有用?？刂苖ap中間結(jié)果是否使用壓縮的參數(shù)為：mapred.compress.map.output(true/false)。將這個(gè)參數(shù)設(shè)置為true時(shí)，那么map在寫中間結(jié)果時(shí)，就會(huì)將數(shù)據(jù)壓縮后再寫入磁盤，讀結(jié)果時(shí)也會(huì)采用先解壓后讀取數(shù)據(jù)。這樣做的后果就是：寫入磁盤的中間結(jié)果數(shù)據(jù)量會(huì)變少，但是cpu會(huì)消耗一些用來壓縮和解壓。所以這種方式通常適合job中間結(jié)果非常大，瓶頸不在cpu，而是在磁盤的讀寫的情況。說的直白一些就是用cpu換IO。根據(jù)觀察，通常大部分的作業(yè)cpu都不是瓶頸，除非運(yùn)算邏輯異常復(fù)雜。所以對(duì)中間結(jié)果采用壓縮通常來說是有收益的。以下是一個(gè)wordcount中間結(jié)果采用壓縮和不采用壓縮產(chǎn)生的map中間結(jié)果本地磁盤讀寫的數(shù)據(jù)量對(duì)比：

map中間結(jié)果不壓縮：

map中間結(jié)果壓縮：

可以看出，同樣的job，同樣的數(shù)據(jù)，在采用壓縮的情況下，map中間結(jié)果能縮小將近10倍，如果map的瓶頸在磁盤，那么job的性能提升將會(huì)非常可觀。

當(dāng)采用map中間結(jié)果壓縮的情況下，用戶還可以選擇壓縮時(shí)采用哪種壓縮格式進(jìn)行壓縮，現(xiàn)在hadoop支持的壓縮格式有：GzipCodec，LzoCodec，BZip2Codec，LzmaCodec等壓縮格式。通常來說，想要達(dá)到比較平衡的cpu和磁盤壓縮比，LzoCodec比較適合。但也要取決于job的具體情況。用戶若想要自行選擇中間結(jié)果的壓縮算法，可以設(shè)置配置參數(shù)：mapred.map.output.compression.codec=org.apache.hadoop.io.compress.DefaultCodec或者其他用戶自行選擇的壓縮方式。

1.2 Map side相關(guān)參數(shù)調(diào)優(yōu)

2 Reduce side tuning參數(shù)

2.1 ReduceTask運(yùn)行內(nèi)部原理

reduce的運(yùn)行是分成三個(gè)階段的。分別為copy->sort->reduce。由于job的每一個(gè)map都會(huì)根據(jù)reduce(n)數(shù)將數(shù)據(jù)分成map 輸出結(jié)果分成n個(gè)partition，所以map的中間結(jié)果中是有可能包含每一個(gè)reduce需要處理的部分?jǐn)?shù)據(jù)的。所以，為了優(yōu)化reduce的執(zhí)行時(shí)間，hadoop中是等job的第一個(gè)map結(jié)束后，所有的reduce就開始嘗試從完成的map中下載該reduce對(duì)應(yīng)的partition部分?jǐn)?shù)據(jù)。這個(gè)過程就是通常所說的shuffle，也就是copy過程。

Reduce task在做shuffle時(shí)，實(shí)際上就是從不同的已經(jīng)完成的map上去下載屬于自己這個(gè)reduce的部分?jǐn)?shù)據(jù)，由于map通常有許多個(gè)，所以對(duì)一個(gè)reduce來說，下載也可以是并行的從多個(gè)map下載，這個(gè)并行度是可以調(diào)整的，調(diào)整參數(shù)為：mapred.reduce.parallel.copies（default 5）。默認(rèn)情況下，每個(gè)只會(huì)有5個(gè)并行的下載線程在從map下數(shù)據(jù)，如果一個(gè)時(shí)間段內(nèi)job完成的map有100個(gè)或者更多，那么reduce也最多只能同時(shí)下載5個(gè)map的數(shù)據(jù)，所以這個(gè)參數(shù)比較適合map很多并且完成的比較快的job的情況下調(diào)大，有利于reduce更快的獲取屬于自己部分的數(shù)據(jù)。

reduce的每一個(gè)下載線程在下載某個(gè)map數(shù)據(jù)的時(shí)候，有可能因?yàn)槟莻€(gè)map中間結(jié)果所在機(jī)器發(fā)生錯(cuò)誤，或者中間結(jié)果的文件丟失，或者網(wǎng)絡(luò)瞬斷等等情況，這樣reduce的下載就有可能失敗，所以reduce的下載線程并不會(huì)無休止的等待下去，當(dāng)一定時(shí)間后下載仍然失敗，那么下載線程就會(huì)放棄這次下載，并在隨后嘗試從另外的地方下載（因?yàn)檫@段時(shí)間map可能重跑）。所以reduce下載線程的這個(gè)最大的下載時(shí)間段是可以調(diào)整的，調(diào)整參數(shù)為：mapred.reduce.copy.backoff（default 300秒）。如果集群環(huán)境的網(wǎng)絡(luò)本身是瓶頸，那么用戶可以通過調(diào)大這個(gè)參數(shù)來避免reduce下載線程被誤判為失敗的情況。不過在網(wǎng)絡(luò)環(huán)境比較好的情況下，沒有必要調(diào)整。通常來說專業(yè)的集群網(wǎng)絡(luò)不應(yīng)該有太大問題，所以這個(gè)參數(shù)需要調(diào)整的情況不多。

Reduce將map結(jié)果下載到本地時(shí)，同樣也是需要進(jìn)行merge的，所以io.sort.factor的配置選項(xiàng)同樣會(huì)影響reduce進(jìn)行merge時(shí)的行為，該參數(shù)的詳細(xì)介紹上文已經(jīng)提到，當(dāng)發(fā)現(xiàn)reduce在shuffle階段iowait非常的高的時(shí)候，就有可能通過調(diào)大這個(gè)參數(shù)來加大一次merge時(shí)的并發(fā)吞吐，優(yōu)化reduce效率。

Reduce在shuffle階段對(duì)下載來的map數(shù)據(jù)，并不是立刻就寫入磁盤的，而是會(huì)先緩存在內(nèi)存中，然后當(dāng)使用內(nèi)存達(dá)到一定量的時(shí)候才刷入磁盤。這個(gè)內(nèi)存大小的控制就不像map一樣可以通過io.sort.mb來設(shè)定了，而是通過另外一個(gè)參數(shù)來設(shè)置：mapred.job.shuffle.input.buffer.percent（default 0.7），這個(gè)參數(shù)其實(shí)是一個(gè)百分比，意思是說，shuffile在reduce內(nèi)存中的數(shù)據(jù)最多使用內(nèi)存量為：0.7 × maxHeap of reduce task。也就是說，如果該reduce
task的最大heap使用量（通常通過mapred.child.java.opts來設(shè)置，比如設(shè)置為-Xmx1024m）的一定比例用來緩存數(shù)據(jù)。默認(rèn)情況下，reduce會(huì)使用其heapsize的70%來在內(nèi)存中緩存數(shù)據(jù)。如果reduce的heap由于業(yè)務(wù)原因調(diào)整的比較大，相應(yīng)的緩存大小也會(huì)變大，這也是為什么reduce用來做緩存的參數(shù)是一個(gè)百分比，而不是一個(gè)固定的值了。

假設(shè)mapred.job.shuffle.input.buffer.percent為0.7，reduce task的max heapsize為1G，那么用來做下載數(shù)據(jù)緩存的內(nèi)存就為大概700MB左右，這700M的內(nèi)存，跟map端一樣，也不是要等到全部寫滿才會(huì)往磁盤刷的，而是當(dāng)這700M中被使用到了一定的限度（通常是一個(gè)百分比），就會(huì)開始往磁盤刷。這個(gè)限度閾值也是可以通過job參數(shù)來設(shè)定的，設(shè)定參數(shù)為：mapred.job.shuffle.merge.percent（default 0.66）。如果下載速度很快，很容易就把內(nèi)存緩存撐大，那么調(diào)整一下這個(gè)參數(shù)有可能會(huì)對(duì)reduce的性能有所幫助。

當(dāng)reduce將所有的map上對(duì)應(yīng)自己partition的數(shù)據(jù)下載完成后，就會(huì)開始真正的reduce計(jì)算階段（中間有個(gè)sort階段通常時(shí)間非常短，幾秒鐘就完成了，因?yàn)檎麄€(gè)下載階段就已經(jīng)是邊下載邊sort，然后邊merge的）。當(dāng)reduce task真正進(jìn)入reduce函數(shù)的計(jì)算階段的時(shí)候，有一個(gè)參數(shù)也是可以調(diào)整reduce的計(jì)算行為。也就是：mapred.job.reduce.input.buffer.percent（default 0.0）。由于reduce計(jì)算時(shí)肯定也是需要消耗內(nèi)存的，而在讀取reduce需要的數(shù)據(jù)時(shí)，同樣是需要內(nèi)存作為buffer，這個(gè)參數(shù)是控制，需要多少的內(nèi)存百分比來作為reduce讀已經(jīng)sort好的數(shù)據(jù)的buffer百分比。默認(rèn)情況下為0，也就是說，默認(rèn)情況下，reduce是全部從磁盤開始讀處理數(shù)據(jù)。如果這個(gè)參數(shù)大于0，那么就會(huì)有一定量的數(shù)據(jù)被緩存在內(nèi)存并輸送給reduce，當(dāng)reduce計(jì)算邏輯消耗內(nèi)存很小時(shí)，可以分一部分內(nèi)存用來緩存數(shù)據(jù)，反正reduce的內(nèi)存閑著也是閑著。

關(guān)于hadoop作業(yè)調(diào)優(yōu)參數(shù)整理及原理是什么問題的解答就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，如果你還有很多疑惑沒有解開，可以關(guān)注億速云行業(yè)資訊頻道了解更多相關(guān)知識(shí)。