Hive經常使用性能優化方法實踐全面總結

Apache Hive做爲處理大數據量的大數據領域數據建設核心工具，數據量每每不是影響Hive執行效率的核心因素，數據傾斜、job數分配的不合理、磁盤或網絡I/O太高、MapReduce配置的不合理等等纔是影響Hive性能的關鍵。

Hive在執行任務時，一般會將Hive SQL轉化爲MapReduce job進行處理。所以對Hive的調優，除了對Hive語句自己的優化，也要考慮Hive配置項以及MapReduce相關的優化。從更底層思考如何優化性能，而不是僅僅侷限於代碼/SQL的層面。


列裁剪和分區裁剪

Hive在讀數據的時候，只讀取查詢中所須要用到的列，而忽略其它列。例如，如有如下查詢：

SELECT age, name FROM people WHERE age > 30;

 

在實施此項查詢中，people表有3列（age，name，address），Hive只讀取查詢邏輯中真正須要的兩列age、name，而忽略列address；這樣作節省了讀取開銷，中間表存儲開銷和數據整合開銷。

同理，對於Hive分區表的查詢，咱們在寫SQL時，經過指定實際須要的分區，能夠減小沒必要要的分區數據掃描【當Hive表中列不少或者數據量很大時，若是直接使用select * 或者不指定分區，效率會很低下（全列掃描和全表掃描）】。

Hive中與列裁剪和分區裁剪優化相關的配置參數分別爲：hive.optimize.cp和hive.optimize.pruner，默認都是true。

 

謂詞下推

在關係型數據庫如MySQL中，也有謂詞下推（Predicate Pushdown，PPD）的概念。它就是將SQL語句中的where謂詞邏輯都儘量提早執行，減小下游處理的數據量。

以下Hive SQL語句：

select    a.*,    b.* from     a join b on (a.id = b.id)where a.id > 15 and b.num > 16;

 

若是沒有謂詞下推，上述SQL須要在完成join處理以後纔會執行where條件過濾。在這種狀況下，參與join的數據可能會很是多，從而影響執行效率。

使用謂詞下推，那麼where條件會在join以前被處理，參與join的數據量減小，提高效率。

在Hive中，能夠經過將參數hive.optimize.ppd設置爲true，啓用謂詞下推。與它對應的邏輯優化器是PredicatePushDown。該優化器就是將OperatorTree中的FilterOperator向上提，見下圖：

 

Hive join優化

關於Hive join，參考文章：《Hive join優化》。

 

hive.fetch.task.conversion

雖然Hive底層能夠將Hive SQL轉化爲MapReduce執行，但有些狀況不使用MapReduce處理效率跟高。好比對於以下SQL：

SELECT name FROM people;

 

在這種狀況下，Hive能夠簡單地讀取people對應的存儲目錄下的文件，而後返回數據。

在hive-default.xml.template文件中hive.fetch.task.conversion默認是more，老版本hive默認是minimal，該屬性修改成more之後，在全局查找、字段查找、limit查找等都不走mapreduce。

<property>
  <name>hive.fetch.task.conversion</name>
  <value>more</value>
  <description>
    Expects one of [none, minimal, more].
    Some select queries can be converted to single FETCH task minimizing latency.
    Currently the query should be single sourced not having any subquery and should not have
    any aggregations or distincts (which incurs RS), lateral views and joins.
    0. none : disable hive.fetch.task.conversion
    1. minimal : SELECT STAR, FILTER on partition columns, LIMIT only
    2. more    : SELECT, FILTER, LIMIT only (support TABLESAMPLE and virtual columns)
  </description>
</property>

 

將hive.fetch.task.conversion設置成none，在Hive shell中執行以下語句，都會執行MapReduce程序。

hive> set hive.fetch.task.conversion=none;
hive> select name from people;
hive> select * from people;

 

把hive.fetch.task.conversion設置成more，而後執行以下語句，以下查詢方式都不會執行MapReduce程序。

hive> set hive.fetch.task.conversion=more;
hive> select name from people;
hive> select * from people;

 

經過參數說明發現當把hive.fetch.task.conversion設置成none時，全部的程序都走mapreduce程序會耗費必定的時間。但就算設置成more，也只有部分sql語句會不走MapReduce程序，那有沒有什麼辦法能夠優化這個問題呢？這就不得不提本地模式了。

 

group by

1）map端預聚合

經過在map端進行一次預聚合（起一個combiner），能夠有效減小shuffle的數據量，而後再在reduce端獲得最終結果。
預聚合的配置參數爲hive.map.aggr，默認值true。

此外，經過hive.groupby.mapaggr.checkinterval參數能夠設置map端預聚合的條數閾值，超過該值就會分拆job，默認值100000。

2）數據傾斜時進行負載均衡處理

當group by時，若是某些key對應的數據量過大，會致使數據傾斜。

經過將參數hive.groupby.skewindata（默認false）設置爲true，那麼在進行group by時，會啓動兩個MR job。第一個job會將map端數據隨機輸入reducer，每一個reducer作部分聚合操做，相同的group by key會分佈在不一樣的reducer中。第二個job再將前面預處理過的數據按key聚合並輸出結果，這樣就起到了均衡的效果。

可是，相對於正常的任務執行，該參數配置爲true時會多啓動一個MR job，會增長開銷，單純依賴它解決數據傾斜並不能從根本上解決問題。所以，建議分析數據、Hive SQL語句等，瞭解產生數據傾斜的根本緣由進行解決。

count(distinct)

count(distinct)採用很是少的reducer進行數據處理。數據量小時對執行效率影響不明顯，可是當數據量大時，效率會很低，尤爲是數據傾斜的時候。

能夠經過group by代替count(distinct)使用。示例以下：

原SQL：SELECT count(DISTINCT id) FROM people;
group by替換後：SELECT count(id) FROM (SELECT id FROM people GROUP BY id) tmp;

 

注意：上述group by替換後，會啓動兩個MR job（只是distinct只會啓動一個），因此要確保啓動job的開銷遠小於計算耗時，才考慮這種方法。不然當數據集很小或者key的傾斜不明顯時，group by還可能會比count（distinct）還慢。

此外，如何用group by方式同時統計多個列？下面提供一種SQL方案：

select tmp.a, sum(tmp.b), count(tmp.c), count(tmp.d) from (
  select a, b, null c, null d from some_table
  union all
  select a, 0 b, c, null d from some_table group by a,c
  union all
  select a, 0 b, null c, d from some_table group by a,d
) tmp;

 

 

笛卡爾積

除非業務須要，在生產中要極力避免笛卡爾積，好比在join語句中不指定on鏈接條件，或者無效的on鏈接條件，Hive只能使用1個reducer來完成笛卡爾積。

 

本地模式

對於處理小數據量的任務，咱們不須要經過集羣模式進行處理（由於爲該任務實際觸發的job執行等開銷可能比實際任務的執行時間還要長），Hive能夠經過本地模式在單臺機器上處理全部的任務。

能夠經過設置hive.exec.mode.local.auto的值爲true，來讓Hive在適當的時候自動啓動這個優化。

 set hive.exec.mode.local.auto=true;

 

設置本地MR的最大輸入數據量，當輸入數據量小於這個值時採用本地MR的方式，默認爲134217728，即128M

set hive.exec.mode.local.auto.inputbytes.max=51234560;

 

設置本地MR的最大輸入文件個數，當輸入文件個數小於這個值時採用本地MR的方式，默認爲4

 set hive.exec.mode.local.auto.input.files.max=10;

 

left semi join替代in/exsits

left semi join是in、exists的高效實現。好比，對於以下SQL

select t1.id, t1.name from t1 where t1.id in (select t2.id from t2);

 

改成left semi join執行：

select t1.id, t1.name from t1 left semi join t2 on t1.id = t2.id;

 

MapReduce相關的優化

1. mapper和reducer個數

關於MapReduce中mapper和reducer個數的決定機制，建議閱讀文章：《詳解MapReduce》

2. 合併小文件

1）輸入階段合併

設置參數hive.input.format爲org.apache.hadoop.hive.ql.io.CombineHiveInputFormat。（默認值是org.apache.hadoop.hive.ql.io.HiveInputFormat）。
此外還需配置兩個參數：mapred.min.split.size.per.node（單節點上的最小split大小）和mapred.min.split.size.per.rack（單機架上的最小split大小）。

若是有split大小小於這兩個值，則會進行合併。

2）輸出階段合併
將hive.merge.mapfiles和hive.merge.mapredfiles都設爲true，前者表示將map-only任務的輸出合併，後者表示將map-reduce任務的輸出合併。

此外，hive.merge.size.per.task能夠指定每一個task輸出後合併文件大小的指望值，hive.merge.size.smallfiles.avgsize能夠指定全部輸出文件大小的均值閾值。若是平均大小不足的話，就會另外啓動一個任務來進行合併。

3. 啓用壓縮

壓縮job的中間結果數據和輸出數據，能夠用少許CPU時間節省不少空間，壓縮方式通常選擇Snappy。（關於Hadoop支持的壓縮格式，參考文章：《 Hadoop支持的壓縮格式對比和應用場景以及Hadoop native庫 》）

要啓用中間壓縮，須要設定hive.exec.compress.intermediate爲true，同時指定壓縮方式hive.intermediate.compression.codec爲org.apache.hadoop.io.compress.SnappyCodec。

另外，參數hive.intermediate.compression.type能夠選擇對塊（BLOCK）仍是記錄（RECORD）壓縮，BLOCK的壓縮率比較高。

輸出壓縮的配置基本相同，打開hive.exec.compress.output便可。

4. JVM重用

在MR job中，默認是每執行一個task就啓動一個JVM。能夠經過配置參數mapred.job.reuse.jvm.num.tasks來進行JVM重用。

例如將這個參數設成5，那麼就表明同一個MR job中順序執行的5個task能夠重複使用一個JVM，減小啓動和關閉的開銷。但它對不一樣MR job中的task無效。

採用合適的存儲格式

在HiveQL的create table語句中，可使用stored as ...指定表的存儲格式。Hive目前支持的存儲格式有TextFile、SequenceFile、RCFile、avro、orc、parquet等。

固然，咱們也能夠採用alter table … [PARTITION partition_spec] set fileformat，修改具體表的文件格式

parquet和orc是企業中經常使用的兩種數據存儲格式，具體能夠參考官網：
https://parquet.apache.org/和https://orc.apache.org/。

 

推測執行

在分佈式集羣環境下，因爲負載不均衡或者資源分佈不均等緣由，會形成同一個做業的多個job之間運行速度不一致，有些job的運行速度可能明顯慢於其餘任務，則這些job會拖慢整個做業的執行進度。爲了不這種狀況發生，Hadoop採用了推測執行（Speculative Execution）機制。

"推測執行"機制，根據必定的規則推測出"拖後腿"的任務，併爲這樣的任務啓動一個備份任務，讓該任務與原始任務同時處理同一份數據，並最終選用最早成功運行完成任務的計算結果做爲最終結果。

Hive一樣能夠開啓推測執行。設置開啓推測執行參數（在配置文件mapred-site.xml中進行配置）

<property>
  <name>mapreduce.map.speculative</name>
  <value>true</value>
  <description>If true, then multiple instances of some map tasks
               may be executed in parallel.</description>
</property>

<property>
  <name>mapreduce.reduce.speculative</name>
  <value>true</value>
  <description>If true, then multiple instances of some reduce tasks
               may be executed in parallel.</description>
</property>

 

hive自己也提供了配置項來控制reduce-side的推測執行：

<property>
    <name>hive.mapred.reduce.tasks.speculative.execution</name>
    <value>true</value>
    <description>Whether speculative execution for reducers should be turned on. </description>
  </property>

 

關於調優這些推測執行變量，目前還很難給出一個具體建議。若是用戶對於運行時的誤差很是敏感的話，那麼能夠將這些功能關閉掉。若是用戶由於輸入數據量很大而須要執行長時間的map或者reduce task的話，那麼啓動推測執行形成的浪費是很是巨大。

 

推薦文章：
一次Java內存泄漏的排查

經典的SparkSQL/Hive-SQL/MySQL面試-練習題

基於Hive進行數倉建設的資源元數據信息統計

Hive Query生命週期 —— 鉤子（Hook）函數篇

Hive實現自增序列及元數據問題

數據倉庫架構和建設方法論

---

關注微信公衆號：大數據學習與分享，獲取更對技術乾貨