Hive查詢生成多個map reduce job,一個map reduce job又有map,reduce,spill,shuffle,sort等多個階段,所以針對hive查詢的優(yōu)化可以大致分為針對MR中單個步驟的優(yōu)化,針對MR全局的優(yōu)化以及針對整個查詢的優(yōu)化。
一個Hive查詢生成多個Map Reduce Job,一個Map Reduce Job又有Map,Reduce,Spill,Shuffle,Sort等多個階段,所以針對Hive查詢的優(yōu)化可以大致分為針對MR中單個步驟的優(yōu)化(其中又會有細分),針對MR全局的優(yōu)化,和針對整個查詢(多MR Job)的優(yōu)化,下文會分別闡述。
在開始之前,先把MR的流程圖帖出來(摘自Hadoop權(quán)威指南),方便后面對照。另外要說明的是,這個優(yōu)化只是針對Hive 0.9版本,而不是后來Hortonwork發(fā)起Stinger項目之后的版本。相對應(yīng)的Hadoop版本是1.x而非2.x。
Map階段的優(yōu)化(Map phase)
Map階段的優(yōu)化,主要是確定合適的Map數(shù)。那么首先要了解Map數(shù)的計算公式:
- Mapred.min.split.size指的是數(shù)據(jù)的最小分割單元大小。
- Mapred.max.split.size指的是數(shù)據(jù)的最大分割單元大小。
- dfs.block.size指的是HDFS設(shè)置的數(shù)據(jù)塊大小。
一般來說dfs.block.size這個值是一個已經(jīng)指定好的值,而且這個參數(shù)Hive是識別不到的:
所以實際上只有Mapred.min.split.size和Mapred.max.split.size這兩個參數(shù)(本節(jié)內(nèi)容后面就以min和max指代這兩個參數(shù))來決定Map數(shù)量。在Hive中min的默認(rèn)值是1B,max的默認(rèn)值是256MB:
所以如果不做修改的話,就是1個Map task處理256MB數(shù)據(jù),我們就以調(diào)整max為主。通過調(diào)整max可以起到調(diào)整Map數(shù)的作用,減小max可以增加Map數(shù),增大max可以減少Map數(shù)。需要提醒的是,直接調(diào)整Mapred.Map.tasks這個參數(shù)是沒有效果的。
調(diào)整大小的時機根據(jù)查詢的不同而不同,總的來講可以通過觀察Map task的完成時間來確定是否需要增加Map資源。如果Map task的完成時間都是接近1分鐘,甚至幾分鐘了,那么往往增加Map數(shù)量,使得每個Map task處理的數(shù)據(jù)量減少,能夠讓Map task更快完成;而如果Map task的運行時間已經(jīng)很少了,比如10-20秒,這個時候增加Map不太可能讓Map task更快完成,反而可能因為Map需要的初始化時間反而讓Job總體速度變慢,這個時候反而需要考慮是否可以把Map的數(shù)量減少,這樣可以節(jié)省更多資源給其他Job。
Reduce階段的優(yōu)化(Reduce phase)
這里說的Reduce階段,是指前面流程圖中的Reduce phase(實際的Reduce計算)而非圖中整個Reduce task。Reduce階段優(yōu)化的主要工作也是選擇合適的Reduce task數(shù)量,跟上面的Map優(yōu)化類似。
與Map優(yōu)化不同的是,Reduce優(yōu)化時,可以直接設(shè)置Mapred。Reduce。tasks參數(shù)從而直接指定Reduce的個數(shù)。當(dāng)然直接指定Reduce個數(shù)雖然比較方便,但是不利于自動擴展。Reduce數(shù)的設(shè)置雖然相較Map更靈活,但是也可以像Map一樣設(shè)定一個自動生成規(guī)則,這樣運行定時Job的時候就不用擔(dān)心原來設(shè)置的固定Reduce數(shù)會由于數(shù)據(jù)量的變化而不合適。
Hive估算Reduce數(shù)量的時候,使用的是下面的公式:
也就是說,根據(jù)輸入的數(shù)據(jù)量大小來決定Reduce的個數(shù),默認(rèn)Hive.exec.Reducers.bytes.per.Reducer為1G,而且Reduce個數(shù)不能超過一個上限參數(shù)值,這個參數(shù)的默認(rèn)取值為999。所以我們可以調(diào)整Hive.exec.Reducers.bytes.per.Reducer來設(shè)置Reduce個數(shù)。
設(shè)置Reduce數(shù)同樣也是根據(jù)運行時間作為參考調(diào)整,并且可以根據(jù)特定的業(yè)務(wù)需求、工作負(fù)載類型總結(jié)出經(jīng)驗,所以不再贅述。
Map與Reduce之間的優(yōu)化(Spill, copy, Sort phase)
Map phase和Reduce phase之間主要有3道工序。首先要把Map輸出的結(jié)果進行排序后做成中間文件,其次這個中間文件就能分發(fā)到各個Reduce,最后Reduce端在執(zhí)行Reduce phase之前把收集到的排序子文件合并成一個排序文件。這個部分可以調(diào)的參數(shù)挺多,但是一般都是不要調(diào)整的,不必重點關(guān)注。
Spill 與 Sort
在Spill階段,由于內(nèi)存不夠,數(shù)據(jù)可能沒辦法在內(nèi)存中一次性排序完成,那么就只能把局部排序的文件先保存到磁盤上,這個動作叫Spill,然后Spill出來的多個文件可以在最后進行merge。如果發(fā)生Spill,可以通過設(shè)置io.Sort.mb來增大Mapper輸出buffer的大小,避免Spill的發(fā)生。另外合并時可以通過設(shè)置io.Sort.factor來使得一次性能夠合并更多的數(shù)據(jù)。調(diào)試參數(shù)的時候,一個要看Spill的時間成本,一個要看merge的時間成本,還需要注意不要撐爆內(nèi)存(io.Sort.mb是算在Map的內(nèi)存里面的)。Reduce端的merge也是一樣可以用io.Sort.factor。一般情況下這兩個參數(shù)很少需要調(diào)整,除非很明確知道這個地方是瓶頸。
Copy
copy階段是把文件從Map端copy到Reduce端。默認(rèn)情況下在5%的Map完成的情況下Reduce就開始啟動copy,這個有時候是很浪費資源的,因為Reduce一旦啟動就被占用,一直等到Map全部完成,收集到所有數(shù)據(jù)才可以進行后面的動作,所以我們可以等比較多的Map完成之后再啟動Reduce流程,這個比例可以通Mapred.Reduce.slowstart.completed.Maps去調(diào)整,他的默認(rèn)值就是5%。如果覺得這么做會減慢Reduce端copy的進度,可以把copy過程的線程增大。tasktracker.http.threads可以決定作為server端的Map用于提供數(shù)據(jù)傳輸服務(wù)的線程,Mapred.Reduce.parallel.copies可以決定作為client端的Reduce同時從Map端拉取數(shù)據(jù)的并行度(一次同時從多少個Map拉數(shù)據(jù)),修改參數(shù)的時候這兩個注意協(xié)調(diào)一下,server端能處理client端的請求即可。
文件格式的優(yōu)化
文件格式方面有兩個問題,一個是給輸入和輸出選擇合適的文件格式,另一個則是小文件問題。小文件問題在目前的Hive環(huán)境下已經(jīng)得到了比較好的解決,Hive的默認(rèn)配置中就可以在小文件輸入時自動把多個文件合并給1個Map處理,輸出時如果文件很小也會進行一輪單獨的合并,所以這里就不專門討論了。相關(guān)的參數(shù)可以在這里找到。
關(guān)于文件格式,Hive0.9版本有3種,textfile,sequencefile和rcfile??傮w上來說,rcfile的壓縮比例和查詢時間稍好一點,所以推薦使用。
關(guān)于使用方法,可以在建表結(jié)構(gòu)時可以指定格式,然后指定壓縮插入:
另外時也可以指定輸出格式,也可以通過Hive。default。fileformat來設(shè)定輸出格式,適用于create table as select的情況:
上面的文件格式轉(zhuǎn)換,其實是由Hive完成的(也就是插入動作)。但是也可以由外部直接導(dǎo)入純文本(可以按照這里的做法預(yù)先壓縮),或者是由MapReduce Job生成的數(shù)據(jù)。
值得注意的是,Hive讀取sequencefile的時候,是把key忽略的,也就是直接讀value并且按照指定分隔符分隔字段。但是如果Hive的數(shù)據(jù)來源是從mr生成的,那么寫sequencefile的時候,key和value都是有意義的,key不能被忽略,而是應(yīng)該當(dāng)成第一個字段。為了解決這種不匹配的情況,有兩種辦法。一種是要求凡是結(jié)果會給Hive用的mr Job輸出value的時候帶上key。但是這樣的話對于開發(fā)是一個負(fù)擔(dān),讀寫數(shù)據(jù)的時候都要注意這個情況。所以更好的方法是第二種,也就是把這個源自于Hive的問題交給Hive解決,寫一個InputFormat包裝一下,把value輸出加上key即可。以下是核心代碼,修改了RecordReader的next方法:
Job整體優(yōu)化
有一些問題必須從Job的整體角度去觀察。這里討論幾個問題:Job執(zhí)行模式(本地執(zhí)行v.s.分布式執(zhí)行)、JVM重用、索引、Join算法、數(shù)據(jù)傾斜。
Job執(zhí)行模式
Hadoop的Map Reduce Job可以有3種模式執(zhí)行,即本地模式,偽分布式,還有真正的分布式。本地模式和偽分布式都是在最初學(xué)習(xí)Hadoop的時候往往被說成是做單機開發(fā)的時候用到。但是實際上對于處理數(shù)據(jù)量非常小的Job,直接啟動分布式Job會消耗大量資源,而真正執(zhí)行計算的時間反而非常少。這個時候就應(yīng)該使用本地模式執(zhí)行mr Job,這樣執(zhí)行的時候不會啟動分布式Job,執(zhí)行速度就會快很多。比如一般來說啟動分布式Job,無論多小的數(shù)據(jù)量,執(zhí)行時間一般不會少于20s,而使用本地mr模式,10秒左右就能出結(jié)果。
設(shè)置執(zhí)行模式的主要參數(shù)有三個,一個是Hive.exec.mode.local.auto,把他設(shè)為true就能夠自動開啟local mr模式。但是這還不足以啟動local mr,輸入的文件數(shù)量和數(shù)據(jù)量大小必須要控制,這兩個參數(shù)分別為Hive.exec.mode.local.auto.tasks.max和Hive.exec.mode.local.auto.inputbytes.max,默認(rèn)值分別為4和128MB,即默認(rèn)情況下,Map處理的文件數(shù)不超過4個并且總大小小于128MB就啟用local mr模式。
JVM重用
正常情況下,MapReduce啟動的JVM在完成一個task之后就退出了,但是如果任務(wù)花費時間很短,又要多次啟動JVM的情況下(比如對很大數(shù)據(jù)量進行計數(shù)操作),JVM的啟動時間就會變成一個比較大的overhead。在這種情況下,可以使用jvm重用的參數(shù):
他的作用是讓一個jvm運行多次任務(wù)之后再退出。這樣一來也能節(jié)約不少JVM啟動時間。
索引
總體上來說,Hive的索引目前還是一個不太適合使用的東西,這里只是考慮到敘述完整性,對其進行基本的介紹。
Hive中的索引架構(gòu)開放了一個接口,允許你根據(jù)這個接口去實現(xiàn)自己的索引。目前Hive自己有一個參考的索引實現(xiàn)(CompactIndex),后來在0.8版本中又加入位圖索引。這里就講講CompactIndex。
CompactIndex的實現(xiàn)原理類似一個lookup table,而非傳統(tǒng)數(shù)據(jù)庫中的B樹。如果你對table A的col1做了索引,索引文件本身就是一個table,這個table會有3列,分別是col1的枚舉值,每個值對應(yīng)的數(shù)據(jù)文件位置,以及在這個文件位置中的偏移量。通過這種方式,可以減少你查詢的數(shù)據(jù)量(偏移量可以告訴你從哪個位置開始找,自然只需要定位到相應(yīng)的block),起到減少資源消耗的作用。但是就其性能來說,并沒有很大的改善,很可能還不如構(gòu)建索引需要花的時間。所以在集群資源充足的情況下,沒有太大必要考慮索引。
CompactIndex的還有一個缺點就是使用起來不友好,索引建完之后,使用之前還需要根據(jù)查詢條件做一個同樣剪裁才能使用,索引的內(nèi)部結(jié)構(gòu)完全暴露,而且還要花費額外的時間。具體看看下面的使用方法就了解了:
Join算法
處理分布式j(luò)oin,一般有兩種方法:
- replication join:把其中一個表復(fù)制到所有節(jié)點,這樣另一個表在每個節(jié)點上面的分片就可以跟這個完整的表join了;
- repartition join:把兩份數(shù)據(jù)按照join key進行hash重分布,讓每個節(jié)點處理hash值相同的join key數(shù)據(jù),也就是做局部的join。
這兩種方式在M/R Job中分別對應(yīng)了Map side join和Reduce side join。在一些MPP DB中,數(shù)據(jù)可以按照某列字段預(yù)先進行hash分布,這樣在跟這個表以這個字段為join key進行join的時候,該表肯定不需要做數(shù)據(jù)重分布了,這種功能是以HDFS作為底層文件系統(tǒng)的Hive所沒有的。
在默認(rèn)情況下,Hive的join策略是進行Reduce side join。當(dāng)兩個表中有一個是小表的時候,就可以考慮用Map join了,因為小表復(fù)制的代價會好過大表Shuffle的代價。使用Map join的配置方法有兩種,一種直接在sql中寫hint,語法是/*+MapJOIN (tbl)*/,其中tbl就是你想要做replication的表。另一種方法是設(shè)置Hive.auto.convert.join = true,這樣Hive會自動判斷當(dāng)前的join操作是否合適做Map join,主要是找join的兩個表中有沒有小表。至于多大的表算小表,則是由Hive.smalltable.filesize決定,默認(rèn)25MB。
但是有的時候,沒有一個表足夠小到能夠放進內(nèi)存,但是還是想用Map join怎么辦?這個時候就要用到bucket Map join。其方法是兩個join表在join key上都做hash bucket,并且把你打算復(fù)制的那個(相對)小表的bucket數(shù)設(shè)置為大表的倍數(shù)。這樣數(shù)據(jù)就會按照join key做hash bucket。小表依然復(fù)制到所有節(jié)點,Map join的時候,小表的每一組bucket加載成hashtable,與對應(yīng)的一個大表bucket做局部join,這樣每次只需要加載部分hashtable就可以了。
然后在兩個表的join key都具有唯一性的時候(也就是可做主鍵),還可以進一步做Sort merge bucket Map join。做法還是兩邊要做hash bucket,而且每個bucket內(nèi)部要進行排序。這樣一來當(dāng)兩邊bucket要做局部join的時候,只需要用類似merge Sort算法中的merge操作一樣把兩個bucket順序遍歷一遍即可完成,這樣甚至都不用把一個bucket完整的加載成hashtable,這對性能的提升會有很大幫助。
然后這里以一個完整的實驗說明這幾種join算法如何操作。
首先建表要帶上bucket:
然后插入我們準(zhǔn)備好的800萬行數(shù)據(jù),注意要強制劃分成bucket(也就是用Reduce劃分hash值相同的數(shù)據(jù)到相同的文件):
這樣這個表就有了800萬id值(且里面沒有重復(fù)值,所以可以做Sort merge),占用80MB左右。
接下來我們就可以一一嘗試Map join的算法了。首先是普通的Map join:
然后就會看到分發(fā)hash table的過程:
不幸的是,居然內(nèi)存不夠了,直接做Map join失敗了。但是80MB的大小為何用1G的heap size都放不下?觀察整個過程就會發(fā)現(xiàn),平均一條記錄需要用到200字節(jié)的存儲空間,這個overhead太大了,對于Map join的小表size一定要好好評估,如果有幾十萬記錄數(shù)就要小心了。雖然不太清楚其中的構(gòu)造原理,但是在互聯(lián)網(wǎng)上也能找到其他的例證,比如這里和這里,平均一行500字節(jié)左右。這個明顯比一般的表一行占用的數(shù)據(jù)量要大。不過Hive也在做這方面的改進,爭取縮小hash table,比如Hive-6430。
所以接下來我們就用bucket Map join,之前分的bucket就派上用處了。只需要在上述sql的前面加上如下的設(shè)置:
然后還是會看到hash table分發(fā):
這次就會看到每次構(gòu)建完一個hash table(也就是所對應(yīng)的對應(yīng)一個bucket),會把這個hash table寫入文件,重新構(gòu)建新的hash table。這樣一來由于每個hash table的量比較小,也就不會有內(nèi)存不足的問題,整個sql也能成功運行。不過光光是這個復(fù)制動作就要花去3分半的時間,所以如果整個Job本來就花不了多少時間的,那這個時間就不可小視。
最后我們試試Sort merge bucket Map join,在bucket Map join的基礎(chǔ)上加上下面的設(shè)置即可:
Sort merge bucket Map join是不會產(chǎn)生hash table復(fù)制的步驟的,直接開始做實際Map端join操作了,數(shù)據(jù)在join的時候邊做邊讀。跳過復(fù)制的步驟,外加join算法的改進,使得Sort merge bucket Map join的效率要明顯好于bucket Map join。
關(guān)于join的算法雖然有這么些選擇,但是個人覺得,對于日常使用,掌握默認(rèn)的Reduce join和普通的(無bucket)Map join已經(jīng)能解決大多數(shù)問題。如果小表不能完全放內(nèi)存,但是小表相對大表的size量級差別也非常大的時候也可以試試bucket Map join,不過其hash table分發(fā)的過程會浪費不少時間,需要評估下是否能夠比Reduce join更高效。而Sort merge bucket Map join雖然性能不錯,但是把數(shù)據(jù)做成bucket本身也需要時間,另外其發(fā)動條件比較特殊,就是兩邊join key必須都唯一(很多介紹資料中都不提這一點。強調(diào)下必須都是唯一,哪怕只有一個表不唯一,出來的結(jié)果也是錯的。當(dāng)然,其實這點完全可以根據(jù)其算法原理推敲出來)。這樣的場景相對比較少見,“用戶基本表 join 用戶擴展表”以及“用戶今天的數(shù)據(jù)快照 join 用戶昨天的數(shù)據(jù)快照”這類場景可能比較合適。
這里順便說個題外話,在數(shù)據(jù)倉庫中,小表往往是維度表,而小表Map join這件事情其實用udf代替還會更快,因為不用單獨啟動一輪Job,所以這也是一種可選方案。當(dāng)然前提條件是維度表是固定的自然屬性(比如日期),只增加不修改(比如網(wǎng)站的頁面編號)的情況也可以考慮。如果維度有更新,要做緩慢變化維的,當(dāng)然還是維表好維護。至于維表原本的一個主要用途OLAP,以Hive目前的性能是沒法實現(xiàn)的,也就不需要多慮了。
數(shù)據(jù)傾斜
所謂數(shù)據(jù)傾斜,說的是由于數(shù)據(jù)分布不均勻,個別值集中占據(jù)大部分?jǐn)?shù)據(jù)量,加上Hadoop的計算模式,導(dǎo)致計算資源不均勻引起性能下降。下圖就是一個例子:
還是拿網(wǎng)站的訪問日志說事吧。假設(shè)網(wǎng)站訪問日志中會記錄用戶的user_id,并且對于注冊用戶使用其用戶表的user_id,對于非注冊用戶使用一個user_id=0代表。那么鑒于大多數(shù)用戶是非注冊用戶(只看不寫),所以user_id=0占據(jù)了絕大多數(shù)。而如果進行計算的時候如果以user_id作為group by的維度或者是join key,那么個別Reduce會收到比其他Reduce多得多的數(shù)據(jù)——因為它要接收所有user_id=0的記錄進行處理,使得其處理效果會非常差,其他Reduce都跑完很久了它還在運行。
傾斜分成group by造成的傾斜和join造成的傾斜,需要分開看。
group by造成的傾斜有兩個參數(shù)可以解決,一個是Hive.Map.aggr,默認(rèn)值已經(jīng)為true,意思是會做Map端的combiner。所以如果你的group by查詢只是做count(*)的話,其實是看不出傾斜效果的,但是如果你做的是count(distinct),那么還是會看出一點傾斜效果。另一個參數(shù)是Hive.groupby. skewindata。這個參數(shù)的意思是做Reduce操作的時候,拿到的key并不是所有相同值給同一個Reduce,而是隨機分發(fā),然后Reduce做聚合,做完之后再做一輪MR,拿前面聚合過的數(shù)據(jù)再算結(jié)果。所以這個參數(shù)其實跟Hive.Map.aggr做的是類似的事情,只是拿到Reduce端來做,而且要額外啟動一輪Job,所以其實不怎么推薦用,效果不明顯。
如果說要改寫SQL來優(yōu)化的話,可以按照下面這么做:
join造成的傾斜,就比如上面描述的網(wǎng)站訪問日志和用戶表兩個表join:
Hive給出的解決方案叫skew join,其原理把這種user_id = 0的特殊值先不在Reduce端計算掉,而是先寫入hdfs,然后啟動一輪Map join專門做這個特殊值的計算,期望能提高計算這部分值的處理速度。當(dāng)然你要告訴Hive這個join是個skew join,即:
還有要告訴Hive如何判斷特殊值,根據(jù)Hive.skewjoin.key設(shè)置的數(shù)量Hive可以知道,比如默認(rèn)值是100000,那么超過100000條記錄的值就是特殊值。
skew join的流程可以用下圖描述:
另外對于特殊值的處理往往跟業(yè)務(wù)有關(guān)系,所以也可以從業(yè)務(wù)角度重寫sql解決。比如前面這種傾斜join,可以把特殊值隔離開來(從業(yè)務(wù)角度說,users表應(yīng)該不存在user_id = 0的情況,但是這里還是假設(shè)有這個值,使得這個寫法更加具有通用性):
數(shù)據(jù)傾斜不僅僅是Hive的問題,其實是share nothing架構(gòu)下必然會碰到的數(shù)據(jù)分布問題,對此學(xué)界也有專門的研究,比如skewtune。
SQL整體優(yōu)化
前面對于單個Job如何做優(yōu)化已經(jīng)做過詳細討論,但是Hive查詢會生成多個Job,針對多個Job,有什么地方需要優(yōu)化?
Job間并行
首先,在Hive生成的多個Job中,在有些情況下Job之間是可以并行的,典型的就是子查詢。當(dāng)需要執(zhí)行多個子查詢union all或者join操作的時候,Job間并行就可以使用了。比如下面的代碼就是一個可以并行的場景示意:
設(shè)置Job間并行的參數(shù)是Hive.exec.parallel,將其設(shè)為true即可。默認(rèn)的并行度為8,也就是最多允許sql中8個Job并行。如果想要更高的并行度,可以通過Hive.exec.parallel. thread.number參數(shù)進行設(shè)置,但要避免設(shè)置過大而占用過多資源。
減少Job數(shù)
另外在實際開發(fā)過程中也發(fā)現(xiàn),一些實現(xiàn)思路會導(dǎo)致生成多余的Job而顯得不夠高效。比如這個需求:查詢某網(wǎng)站日志中訪問過頁面a和頁面b的用戶數(shù)量。低效的思路是面向明細的,先取出看過頁面a的用戶,再取出看過頁面b的用戶,然后取交集,代碼如下:
這樣一來,就要產(chǎn)生2個求子查詢的Job,一個用于關(guān)聯(lián)的Job,還有一個計數(shù)的Job,一共有4個Job。
但是我們直接用面向統(tǒng)計的方法去計算的話(也就是用group by替代join),則會更加符合M/R的模式,而且生成了一個完全不帶子查詢的sql,只需要用一個Job就能跑完:
第一種查詢方法符合思考問題的直覺,是工程師和分析師在實際查數(shù)據(jù)中最先想到的寫法,但是如果在目前Hive的query planner不是那么智能的情況下,想要更加快速的跑出結(jié)果,懂一點工具的內(nèi)部機理也是必須的。(作者:孫逸 / 審校:劉亞瓊)
作者介紹:孫逸,zamplus,數(shù)據(jù)倉庫工程師。從事數(shù)據(jù)倉庫平臺建設(shè)與數(shù)據(jù)產(chǎn)品開發(fā),熱衷于研究大數(shù)據(jù)相關(guān)的計算技術(shù),并應(yīng)用于實踐之中,讓數(shù)據(jù)產(chǎn)生真正價值。
聯(lián)系客服