Apache Hive-通用優化-featch抓取機制 mr本地模式

Fetch抓取機制

• 功能：在執行sql的時候，能不走MapReduce程序處理就盡量不走MapReduce程序處理.
• 盡量直接去操作數據文件。

設置： hive.fetch.task.conversion= more。

--在下述3種情況下 sql不走mr程序
--全局查找
select * from student;
--字段查找
select num,name from student;
--limit 查找
select num,name from student limit 2;

mapreduce本地模式

• MapReduce程序除了可以提交到yarn集群分布式執行之外，還可以使用本地模擬環境運行，當然此時就不是分布式執行的程序，但是針對小文件小數據處理特別有效果。
• 用戶可以通過設置hive.exec.mode.local.auto的值為true，來讓Hive在適當的時候自動啟動這個 優化。

功能：如果非要執行==MapReduce程序，能夠本地執行的，盡量不提交yarn上執行==。

默認是關閉的。意味著只要走MapReduce就提交yarn執行。

mapreduce.framework.name = local 本地模式
mapreduce.framework.name = yarn 集群模式 

Hive提供了一個參數，自動切換MapReduce程序為本地模式，如果不滿足條件，就執行yarn模式。

set hive.exec.mode.local.auto = true;
--3個條件必須都滿足 自動切換本地模式
The total input size of the job is lower than: hive.exec.mode.local.auto.inputbytes.max (128MB by default)  --數據量小于128M
The total number of map-tasks is less than: hive.exec.mode.local.auto.tasks.max (4 by default)  --maptask個數少于4個
The total number of reduce tasks required is 1 or 0.  --reducetask個數是0 或者 1

切換Hive的執行引擎

WARNING: Hive-on-MR is deprecated in Hive 2 and may not be available in the future versions. Consider using a different execution engine (i.e. spark, tez) or using Hive 1.X releases.

如果針對Hive的調優依然無法滿足你的需求 還是效率低， 嘗試使用spark計算引擎 或者Tez.

Apache Hive-通用優化-join優化

在了解join優化的時候,我們需要了解一個前置知識點:map端join 和reduce端join

- reduce端join

• 這種join的弊端在于map階段沒有承擔太多的責任,所有的數據在經過shuffle在reduce階段實現的,而shuffle又是影響性能的核心點.

-map端join

• 首先啟動本地任務將join中小表數據進行分布式緩存
• 啟動mr程序(只有map階段)并行處理大數據,并且從自己的緩存中讀取小表數據,進行join,結果直接輸出到文件中
• 沒有shuffle過程 也沒有reduce過程
• 弊端:緩存太小導致表數據不能太大

reduce 端 join 優化

適合于大表Join大表

bucket join-- 適合于大表Join大表

方式1：Bucktet Map Join 分桶表

語法: clustered by colName(參與join的字段)
參數: set hive.optimize.bucketmapjoin = true
要求: 分桶字段 = Join字段 ，分桶的個數相等或者成倍數，必須是在map join中

方式2：Sort Merge Bucket Join（SMB）

基于有序的數據Join
語法:clustered by colName sorted by (colName)
參數
? ? set hive.optimize.bucketmapjoin = true;
? ? set hive.auto.convert.sortmerge.join=true;
? ? set hive.optimize.bucketmapjoin.sortedmerge = true;
? ? set hive.auto.convert.sortmerge.join.noconditionaltask=true;
要求: 分桶字段 = Join字段 = 排序字段,分桶的個數相等或者成倍數

map 端 join 優化

• hive.auto.convert.join.noconditionaltask

hive.auto.convert.join=true
Hive老版本
#如果參與的一個表大小滿足條件 轉換為map join
hive.mapjoin.smalltable.filesize=25000000  
Hive2.0之后版本
#是否啟用基于輸入文件的大小，將reduce join轉化為Map join的優化機制。假設參與join的表(或分區)有N個，如果打開這個參數，并且有N-1個表(或分區)的大小總和小于hive.auto.convert.join.noconditionaltask.size參數指定的值，那么會直接將join轉為Map join。
hive.auto.convert.join.noconditionaltask=true 
hive.auto.convert.join.noconditionaltask.size=512000000 

Apache Hive--通用調優--數據傾斜優化

數據傾斜優化
? ? 什么是數據傾斜
? ? ? ? 描述的數據進行分布式處理 ?分配不平均的現象
? ? 數據傾斜的后果
? ? ? ? 某個task數據量過大 執行時間過長 ?導致整體job任務遲遲不結束
? ? ? ? ? ? 執行時間長 ?出bug及風險幾率提高
? ? ? ? ? ? 霸占運算資源 遲遲不釋放
? ? 通常如何發現數據傾斜
? ? ? ? 在yarn或者其他資源監控軟件上 ?發現某個job作業 卡在某個進度遲遲不動 （注意 倒不是報錯）
? ? 造成數據傾斜的原因
? ? ? ? 數據本身就傾斜
? ? ? ? 自定義分區、分組規則不合理
? ? ? ? 業務影響 造成數據短期高頻波動
? ? 數據傾斜的通用解決方案
? ? ? ? 1、有錢 ?有預警 ?
? ? ? ? ? ? 增加物理資源 ?單獨處理傾斜的數據
? ? ? ? 2、沒錢 ?沒有預警
? ? ? ? ? ? 傾斜數據打散 ?分步執行
? ? ? ? ? ? ? ? 先將傾斜數據打散成多干份?
? ? ? ? ? ? ? ? 處理的結果再最終合并
? ? hive中數據傾斜的場景
? ? ? ? 場景一：group by ?、count(distinct)
? ? ? ? ? ? hive.map.aggr=true; ?map端預聚合
? ? ? ? ? ? 手動將數據隨機分區 ?select * from table distribute by rand();
? ? ? ? ? ? 如果有數據傾斜問題 ?開啟負載均衡
? ? ? ? ? ? ? ? 先啟動第一個mr程序 把傾斜的數據隨機打散分散到各個reduce中
? ? ? ? ? ? ? ? 然后第二個mr程序把上一步結果進行最終匯總
? ? ? ? ? ? ? ? hive.groupby.skewindata=true;
? ? ? ? 場景二：join
? ? ? ? ? ? 提前過濾，將大數據變成小數據，實現Map Join
? ? ? ? ? ? 使用Bucket Join
? ? ? ? ? ? 使用Skew Join
? ? ? ? ? ? ? ? 將Map Join和Reduce Join進行合并，如果某個值出現了數據傾斜，就會將產生數據傾斜的數據單獨使用Map Join來實現
? ? ? ? ? ? ? ? 最終將Map Join的結果和Reduce Join的結果進行Union合并
? ? ? ? Hive中通常指的是在reduce階段數據傾斜

解決方法

group by數據傾斜

方案一：開啟Map端聚合

hive.map.aggr=true;
#是否在Hive Group By 查詢中使用map端聚合。
#這個設置可以將頂層的部分聚合操作放在Map階段執行，從而減輕清洗階段數據傳輸和Reduce階段的執行時間，提升總體性能。但是指標不治本。

方案二：實現隨機分區

實現隨機分區
select * from table distribute by rand();

方案三：數據傾斜時==自動負載均衡==只使用group by

hive.groupby.skewindata=true;
#開啟該參數以后，當前程序會自動通過兩個MapReduce來運行
#第一個MapReduce自動進行隨機分布到Reducer中，每個Reducer做部分聚合操作，輸出結果
#第二個MapReduce將上一步聚合的結果再按照業務（group by key）進行處理，保證相同的分布到一起，最終聚合得到結果

join數據傾斜

• 方案一：提前過濾，將大數據變成小數據，實現Map Join
• 方案二：使用Bucket Join
• 方案三：使用Skew Join

數據單獨使用Map Join來實現

#其他沒有產生數據傾斜的數據由Reduce Join來實現，這樣就避免了Reduce Join中產生數據傾斜的問題
#最終將Map Join的結果和Reduce Join的結果進行Union合并
#開啟運行過程中skewjoin
set hive.optimize.skewjoin=true;
#如果這個key的出現的次數超過這個范圍
set hive.skewjoin.key=100000;
#在編譯時判斷是否會產生數據傾斜
set hive.optimize.skewjoin.compiletime=true;
set hive.optimize.union.remove=true;
#如果Hive的底層走的是MapReduce，必須開啟這個屬性，才能實現不合并
set mapreduce.input.fileinputformat.input.dir.recursive=true;

Apache Hive--通用調優--MR程序task個數調整

maptask個數

• 如果是在MapReduce中 maptask是通過==邏輯切片==機制決定的。
• 但是在hive中，影響的因素很多。比如邏輯切片機制，文件是否壓縮、壓縮之后是否支持切割。
• 因此在==Hive中，調整MapTask的個數，直接去HDFS調整文件的大小和個數，效率較高==。

合并的大小最好=block size

如果大文件多，就調整blocl size

reducetask個數

• 如果在MapReduce中，通過代碼可以直接指定 job.setNumReduceTasks(N)
• 在Hive中，reducetask個數受以下幾個條件控制的

hive.exec.reducers.bytes.per.reducer=256000000
每個任務最大的 reduce 數，默認為 1009
hive.exec.reducsers.max=1009
mapreduce.job.reduces
該值默認為-1，由 hive 自己根據任務情況進行判斷。

--如果用戶用戶不設置 hive將會根據數據量或者sql需求自己評估reducetask個數。
--用戶可以自己通過參數設置reducetask的個數
?set mapreduce.job.reduces = N
--用戶設置的不一定生效，如果用戶設置的和sql執行邏輯有沖突，比如order by，在sql編譯期間，hive又會將reducetask設置為合理的個數。 ?

Number of reduce tasks determined at compile time: 1

通用優化-執行計劃

通過執行計劃可以看出==hive接下來是如何打算執行這條sql的==。

語法格式：explain + sql語句

通用優化-并行機制,推測執行機制

并行執行機制

• 如果hivesql的底層某些stage階段可以并行執行，就可以提高執行效率。
• 前提是==stage之間沒有依賴== 并行的弊端是瞬時服務器壓力變大。

參數

set hive.exec.parallel=true; --是否并行執行作業。適用于可以并行運行的 MapReduce 作業，例如在多次插入期間移動文件以插入目標
set hive.exec.parallel.thread.number=16; --最多可以并行執行多少個作業。默認為8。

Hive的嚴格模式

• 注意。不要和動態分區的嚴格模式搞混淆。
• 這里的嚴格模式指的是開啟之后 ==hive會禁止一些用戶都影響不到的錯誤包括效率低下的操作==，不允許運行一些有風險的查詢。

設置

set hive.mapred.mode = strict --默認是嚴格模式  nonstrict

解釋

1、如果是分區表，沒有where進行分區裁剪 禁止執行

2、order by語句必須+limit限制

推測執行機制 ==建議關閉==。

• MapReduce中task的一個機制。
• 功能：

一個job底層可能有多個task執行，如果某些拖后腿的task執行慢，可能會導致最終job失敗。

所謂的==推測執行機制就是通過算法找出拖后腿的task,為其啟動備份的task==。

兩個task同時處理一份數據，誰先處理完，誰的結果作為最終結果。

• 推測執行機制默認是開啟的，但是在企業生產環境中==建議關閉==。