技術文章 » HiveSQL在使用聚合類函數的時候效能分析和優化詳解

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

2023-06-21 12:01:00

概述

前文我們寫過簡單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，隨機贈送一本魯邊備註版珍藏巨量資料書籍。