HiveSQL在使用聚合類函數的時候性能分析和優化詳解

帶聚合函數的SQL邏輯，我們可以根據其執行過程的不同，將其分成三大類來進行分析：僅在Reduce階段聚合的SQL執行邏輯在Map和Reduce階段都有聚合操作的SQL執行邏輯高級分組聚合的執行SQL邏輯 ...

概述

前文我們寫過簡單SQL的性能分析和解讀，簡單SQL被歸類為select-from-where型SQL語句，其主要特點是只有map階段的數據處理，相當於直接從hive中取數出來，不需要經過行變化。在非多個節點的操作上，其性能甚至不比Tez和Spark差。

而這次我們主要說的是使用聚合類函數的hiveSQL，這類SQL需要完整的map階段和reduce階段才能完成數據處理。我們把它可以歸類為select-aggr_function-from-where-groupby 類型SQL語句。

在生產環境中我們一般常用的聚合函數見如下列表：

函數	參數格式	解釋
count	count(*), count(expr)，count(distinct expr)	返回查找的總行數，count(*)返回的行數包括null值；count(expr)和count(distinct expr) 不包括null值
sum	sum(col), sum(DISTINCT col)	sum(col)返回組內查詢列元素的總和，sum(DISTINCT col)返回組內查詢列列的不同值的總和
avg	avg(col), avg(DISTINCT col)	sum(col)返回組內查詢列元素的平均值，sum(DISTINCT col)返回組內查詢列的不同值的平均值
min	min(col)	返回組內查詢列的最小值
max	max(col)	返回組內查詢列的最大值
variance/var_pop	variance(col)/var_pop(col)	返回組內查詢列的方差（也可稱為總體方差），也可寫成var_pop(col)
var_samp	var_samp(col)	返回組內查詢列方差的無偏估計（方差無偏估計中，因為估計期望損失了一個自由度，估計的分母為n-1，也可稱為樣本方差）
stddev_pop	stddev_pop(col)	返回組內查詢列的標準差
stddev_samp	stddev_samp(col)	返回組內查詢列標準差的無偏估計方差（無偏估計中，因為估計期望損失了一個自由度，估計的分母為n-1）
covar_pop	covar_pop(col1, col2)	返回組內查詢列col1和col2的總體協方差
covar_samp	covar_samp(col1, col2)	返回組內查詢列col1和col2的樣本協方差
corr	corr(col1, col2)	返回組內查詢列col1和col2的相關係數
percentile	percentile(BIGINT col, p)	返回組內查詢整數列col所在的分位數，p可以為浮點數或數組，且其中元素大小必須在0-1之間。若col不是整數，需使用percentile_approx
percentile_approx	percentile_approx(DOUBLE col, array(p1[, p2]…) [, B])	返回組內查詢列col所在的分位數，p可以為浮點數或數組，且其中元素大小必須在0-1之間。B為可選參數，為精度控制參數
regr_avgx	regr_avgx(independent, dependent)	計算自變數的平均值。該函數將任意一對數字類型作為參數，並返回一個double。任何具有null的對都將被忽略。如果應用於空集：返回null。否則，它計算以下內容：avg(dependent)
regr_avgy	regr_avgy(independent, dependent)	計算因變數的平均值。該函數將任意一對數字類型作為參數，並返回一個double。任何具有null的對都將被忽略。如果應用於空集：返回null。否則，它計算以下內容：avg(independent)
regr_count	regr_count(independent, dependent)	返回independent和dependent都非空的對數
regr_intercept	regr_intercept(independent, dependent)	返回線性回歸的截距項
regr_r2	regr_r2(independent, dependent)	返回線性回歸的判決繫數（R方，coefficient of determination）
regr_slope	regr_slope(independent, dependent)	返回線性回歸的斜率繫數
regr_sxx	regr_sxx(independent, dependent)	等價於regr_count(independent, dependent) * var_pop(dependent)
regr_sxy	regr_sxy(independent, dependent)	regr_count(independent, dependent) * covar_pop(independent, dependent)
regr_syy	regr_syy(independent, dependent)	regr_count(independent, dependent) * var_pop(independent)
histogram_numeric	histogram_numeric(col, b)	用於畫直方圖。返回一個長度為b的數組，數組中元素為(x,y)形式的鍵值對，x代表了直方圖中該柱形的中心，y代表可其高度。
collect_set	collect_set(col)	返回查詢列col去重後的集合，與distinct不同，distinct查詢結果為一列數據，collect_set查詢後結果為一個集合形式的元素
collect_list	collect_list(col)	返回查詢列col的列表
ntile	ntile(INTEGER x)	將有序分區劃分為x個稱為存儲桶的組，併為該分區中的每一行分配存儲桶編號。（此方式存儲可以快速計算分位數）

對於帶聚合函數的SQL邏輯，我們可以根據其執行過程的不同，將其分成三大類來進行分析：

僅在Reduce階段聚合的SQL執行邏輯
在Map和Reduce階段都有聚合操作的SQL執行邏輯
高級分組聚合的執行SQL邏輯

1.僅在Reduce階段聚合的SQL執行邏輯

我們通過SQL執行計劃來解讀Reduce階段聚合的SQL邏輯，如一下實例：

例1 在Reduce階段進行聚合的SQL邏輯

set hive.map.aggr=false;
explain
-- 小於30歲人群的不同性別平均年齡
select gender,avg(age) as avg_age from temp.user_info_all where ymd = '20230505'
and age < 30 
group by gender;

其執行結果如下內容：

STAGE DEPENDENCIES:
  Stage-1 is a root stage
  Stage-0 depends on stages: Stage-1

STAGE PLANS:
  Stage: Stage-1
    Map Reduce
      Map Operator Tree:
          TableScan
            alias: user_info_all
            Statistics: Num rows: 32634295 Data size: 783223080 Basic stats: COMPLETE Column stats: NONE
            Filter Operator
              predicate: (age < 30) (type: boolean)
              Statistics: Num rows: 10878098 Data size: 261074352 Basic stats: COMPLETE Column stats: NONE
              Reduce Output Operator
                key expressions: gender (type: int)
                sort order: +
                Map-reduce partition columns: gender (type: int)
                Statistics: Num rows: 10878098 Data size: 261074352 Basic stats: COMPLETE Column stats: NONE
                value expressions: age (type: bigint)
      Reduce Operator Tree:
        Group By Operator
          aggregations: avg(VALUE._col0)
          keys: KEY._col0 (type: int)
          mode: complete
          outputColumnNames: _col0, _col1
          Statistics: Num rows: 5439049 Data size: 130537176 Basic stats: COMPLETE Column stats: NONE
          File Output Operator
            compressed: true
            Statistics: Num rows: 5439049 Data size: 130537176 Basic stats: COMPLETE Column stats: NONE
            table:
                input format: org.apache.hadoop.mapred.SequenceFileInputFormat
                output format: org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat
                serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe

  Stage: Stage-0
    Fetch Operator
      limit: -1
      Processor Tree:
        ListSink

以上內容的具體關鍵字就不作解讀了，在Hive執行計劃之一文讀懂Hive執行計劃中已經做了完整的解釋，看不懂請回看。

從上述信息中可以看到Map階段的解析被分解為常規的三大步驟。

TableScan
Filter Operator
Reduce Output Operator

Reduce階段的解析被分解為兩步：

Group By Operator
File Output Operator

對比之前簡單SQL執行步驟過程。

可以直觀看出簡單SQL的執行邏輯主要是在進行列投影後就直接將數據寫入本地。而在聚合函數的SQL執行過程中使用到了Reduce階段，多了輸出到reduce階段和分組聚合操作。

其中從map階段輸出到reduce階段的這個流程，我們稱之為數據的shuffle。後續有機會可以詳細講解其過程。

通過以上案例，可以直觀的看出該SQL邏輯在map階段沒有計算的操作，只是對數據進行了一個重新組織，之後在寫入reduce，即shuffle的過程進行排序，寫記憶體，寫磁碟，然後網路傳輸等工作。這塊如果在map階段的數據量很大，就會占用比較多的資源。

那麼如何進行優化呢？

2.在map和reduce階段聚合的SQL邏輯

以上例1，可以看到我設置了一個參數set hive.map.aggr=false;

該參數我的集群是預設開啟的，為了演示我這裡設置關閉。這參數本身開啟後起到的作用是提前在map階段進行數據彙總，即Combine操作。

map端數據過大一般的優化方式有兩種：

啟用Combine操作，進行提前聚合，進而減少shuffle的數據量，減少資源消耗。
啟用數據壓縮來減少Map和Reduce之間傳輸的數據量。

一般的數據壓縮方式就是我們在hive上使用的數據存儲格式和數據壓縮方法。

啟用Combine操作，在hive中提供了對應的參數，set hive.map.aggr=true;通過該配置可以控制是否啟用Map端的聚合。

可以看如下例子：

例2 啟用Map端聚合的SQL邏輯

同樣的SQL邏輯

set hive.map.aggr=true;
explain
-- 小於30歲人群的不同性別平均年齡
select gender,avg(age) as avg_age from temp.user_info_all where ymd = '20230505'
and age < 30 
group by gender;

其執行計劃結果如下：

STAGE DEPENDENCIES:
  Stage-1 is a root stage
  Stage-0 depends on stages: Stage-1

STAGE PLANS:
  Stage: Stage-1
    Map Reduce
      Map Operator Tree:
          TableScan
            alias: user_info_all
            Statistics: Num rows: 32634295 Data size: 783223080 Basic stats: COMPLETE Column stats: NONE
            Filter Operator
              predicate: (age < 30) (type: boolean)
              Statistics: Num rows: 10878098 Data size: 261074352 Basic stats: COMPLETE Column stats: NONE
              Group By Operator
                aggregations: avg(age)
                keys: gender (type: int)
                mode: hash
                outputColumnNames: _col0, _col1
                Statistics: Num rows: 10878098 Data size: 261074352 Basic stats: COMPLETE Column stats: NONE
                Reduce Output Operator
                  key expressions: _col0 (type: int)
                  sort order: +
                  Map-reduce partition columns: _col0 (type: int)
                  Statistics: Num rows: 10878098 Data size: 261074352 Basic stats: COMPLETE Column stats: NONE
                  value expressions: _col1 (type: struct<count:bigint,sum:double,input:bigint>)
      Reduce Operator Tree:
        Group By Operator
          aggregations: avg(VALUE._col0)
          keys: KEY._col0 (type: int)
          mode: mergepartial
          outputColumnNames: _col0, _col1
          Statistics: Num rows: 5439049 Data size: 130537176 Basic stats: COMPLETE Column stats: NONE
          File Output Operator
            compressed: true
            Statistics: Num rows: 5439049 Data size: 130537176 Basic stats: COMPLETE Column stats: NONE
            table:
                input format: org.apache.hadoop.mapred.SequenceFileInputFormat
                output format: org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat
                serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe

  Stage: Stage-0
    Fetch Operator
      limit: -1
      Processor Tree:
        ListSink

這裡說明一下 value expressions: _col1 (type: struct<count:bigint,sum:double,input:bigint>)

在map階段的最後map端最終輸出的結果為一個結構體struct。其中map階段不能計算平均值，只能計算總數和對應個數，這兩者分別對應結構體中的sum和count。

將以上邏輯進行流程化。

對比例1 操作流程圖，可以看出來例2 在map階段多了一個分組聚合操作。

文字描述：先將本地節點的數據進行一個初步聚合，求出該性別的年齡相加總數和用戶個數。這就已經極大的減少了數據量。之後再進行數據shuffle（分發）過程，將各個節點的數據進行彙總，之後在reduce階段，再進行二次聚合。將各個節點的求和值和計數值彙總。在得到具體的平均值。該計算完成，輸出。

以上，開啟map端聚合，這也是hive在使用聚合函數過程中的最常用的一個優化方式。

hive.map.aggr=true;

那麼，有一個問題，如何解決map端的數據傾斜問題？以下為常規手段。

在mr程式上我們可以說開啟Combine模式，進行map端聚合，hive上我們可以說開啟map端聚合參數。
還有，採用更優的壓縮演算法和數據存儲格式。

思考一下，以上方式其實更多的是提供一個將大量數據變小的方式，那麼map端真正的數據傾斜是什麼造成的，核心該如何處理。

下一期：什麼是hive的高級分組聚合，它的用法和註意事項有哪些

按例，歡迎點擊此處關註我的個人公眾號，交流更多知識。

後臺回覆關鍵字 hive，隨機贈送一本魯邊備註版珍藏大數據書籍。