此次實踐緣由

hive自去年學習後，就一直放著少有使用了，前幾天接到面試，說hive這塊，讓再多學習學習。
所以有了這次實踐，複習一把。
主要是想做些優化，以及嘗試模擬出資料傾斜並解決傾斜的問題

關於實驗

實驗的前半部分大部分參考的林子雨實驗室的案例
林博士團隊的案例，確實講得很詳細，
關於這一部分，我在走一遍流程的前提下，提煉下要素，並對原案例進行一些更深入的觀察。這些內容大都使用黃色背景高亮

我的重心是，hive做mapreduce工作這一塊的優化，以及嘗試模擬出資料傾斜並解決傾斜的問題。

hive、hbase、mysql資料傳輸

這部分，大都來自林子雨實驗室

關於叢集

：Cloudera CDH5.13.2，搭建過程可參考此文

資料集下載：https://pan.baidu.com/s/1nuOSo7B
【說句題外話，用了一年多的破解版網盤，我欠百度網盤一個會員。。。等秋招順利拿了offer】

資料集預覽

mkdir -p /home/guest/bigdatacase/dataset

scp ~/elk/user.zip [email protected]_remote_ipaddress:/home/guest/bigdatacase/dataset

unzip /home/guest/bigdatacase/dataset/user.zip

有兩個檔案：raw_user.csv和small_user.csv。

head /usr/local/bigdatacase/dataset

每行記錄都包含5個欄位，資料集中的欄位及其含義如下：

user_id (使用者id)
item_id (商品id)
behaviour_type (包括瀏覽、收藏、加購物車、購買，對應取值分別是1、2、3、4)
user_geohash (使用者地理位置雜湊值，有些記錄中沒有這個欄位值，後面用指令碼做資料預處理時把這個欄位全部刪除)
item_category (商品分類)
time (該記錄產生時間)

資料預處理

為每行記錄增加一個id欄位（讓記錄具有唯一性）、增加一個省份欄位（用來後續進行視覺化分析），並且丟棄user_geohash欄位（後面分析不需要這個欄位）

指令碼如下

下面的程式碼的基本形式是： awk -F “,” ‘處理邏輯’ $infile > $outfile

#!/bin/bash
infile=$1
outfile=$2
#最後的$infile > $outfile必須跟在}’這兩個字元的後面
awk -F "," 'BEGIN{
        srand();
        id=0;
        Province[0]="山東";Province[1]="山西";Province[2]="河南";Province[3]="河北";Province[4]="陝西";Province[5]="內蒙古";Province[6]="上海市";
        Province[7]="北京市";Province[8]="重慶市";Province[9]="天津市";Province[10]="福建";Province[11]="廣東";Province[12]="廣西";Province[13]="雲南"; 
        Province[14]="浙江";Province[15]="貴州";Province[16]="新疆";Province[17]="西藏";Province[18]="江西";Province[19]="湖南";Province[20]="湖北";
        Province[21]="黑龍江";Province[22]="吉林";Province[23]="遼寧"; Province[24]="江蘇";Province[25]="甘肅";Province[26]="青海";Province[27]="四川";
        Province[28]="安徽"; Province[29]="寧夏";Province[30]="海南";Province[31]="香港";Province[32]="澳門";Province[33]="臺灣";
    }
    {
        id=id+1;
        value=int(rand()*34);       
        print id"\t"$1"\t"$2"\t"$3"\t"$5"\t"substr($6,1,10)"\t"Province[value]
    }' $infile > $outfile

[關於shell指令碼中，=兩邊的空格](https://blog.csdn.net/solarsaber/article/details/65938545)  先用小的資料集走下流程

bash ./pre_deal.sh small_user.csv user_table.txt

處理之後的資料：

匯入資料到hive

補充 匯入資料到hive表的六種方式：
後文優化，可以從這個角度進行部分優化
● 載入本地檔案
● 載入HDFS檔案
● 載入並覆蓋
● 通過查詢到的檔案載入：insert into table tablename select value from tablename
● 建立表時使用select載入
● 指定location載入

這裡建立外部表，所以先得把資料放入hdfs。
其實資料放入外部表本身就已經是一種優化策略了，因為hive資料倉庫的概念，這裡的資料，我們可以考慮為，多個業務部門都需要訪問此資料。所以，在實踐中，我們可以將最大的原始資料放入hdfs，然後子業務根據各自需要，從最大的表中取資料。而該原始表，最好使用外部表——External_Table ，外部表與管理表——Manage_Table最大的區別在於外部表刪除表時，僅僅刪除儲存在關係型資料庫中的元資料，而不會刪除hdfs中資料。

sudo -u hdfs hadoop fs -mkdir /tmp/data21080729
sudo -u hdfs hdfs dfs -put /home/user_table.txt /tmp/data21080729[你也許需要修改檔案許可權

建立外部表

【外部表建立後會自動載入資料】
create database tmp

CREATE EXTERNAL TABLE tmp.bigdata_user(id INT,uid STRING,item_id STRING,behavior_type INT,item_category STRING,visit_date DATE,province STRING) COMMENT 'Welcome to xmu dblab!' ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LOCATION '/tmp/data21080729';

先簡單感受下mapreduce離線的慢：

其實真正的mapreduce時間也並沒有35秒那麼多
namenode，8088埠：

進一步的查詢，以及使用較大的那一份資料集，放在後面，先完成下hive到mysql，mysql到hbase

hive到mysql

額，學校的VPN，連著老是斷，我們先screen走一波
screen -S hmh
screen -ls
這裡我們不在之前的表上操作，我們建立一個臨時表，hive中，臨時表只在當前會話起作用，會話結束，表 自動刪除。
注意，林子雨實驗室的案例，建立的並不是臨時表，缺關鍵字temporary

hive --version
從Hive1.1開始臨時表可以儲存在記憶體或SSD，使用hive.exec.temporary.table.storage引數進行配置，該引數有三種取值：memory、ssd、default。
儲存層次
所以，在記憶體裝得下的情況下【一般臨時表也使用select from的形式建立，所以就這個應該不是大問題】，我們還可以讓臨時表常駐記憶體，加快運算速度，這也可算作優化策略之一。

相關issue:HIVE-7090
set hive.exec.temporary.table.storage=memory;

• 建立臨時表
  create temporary table tmp.user_action(id STRING,uid STRING, item_id STRING, behavior_type STRING, item_category STRING, visit_date DATE, province STRING) COMMENT 'temporary table to mysql use sqoop! ' ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' STORED AS TEXTFILE;


describe formatted tmp.user_action;
可以發現，臨時表是這樣儲存的【不指定location】：

hdfs://ip10:8020/tmp/hive/hdfs/de47d44d-1eb4-411b-bb4f-2b9eaf19842f/_tmp_space.db/4784a44c-95db-43d7-847c-4b05879f6e8a

非臨時表，且不指定location：

/user/hive/warehouse/tmp.db

• 從另外一個表以select的方式載入資料：
  INSERT OVERWRITE TABLE tmp.user_action select * from tmp.bigdata_user;
  觀察列印的資訊發現，這裡跑了三個mapreduce，好奇為什麼會有三個，具體是哪些階段發生了mapreduce，可以使用EXPLAIN命令檢視。
• 用sqoop匯入mysql
  create database hive_to_mysql;

CREATE TABLE `hive_to_mysql`.`user_action` (`id` varchar(50),`uid` varchar(50),`item_id` varchar(50),`behavior_type` varchar(10),`item_category` varchar(50), `visit_date` DATE,`province` varchar(20)) ENGINE=InnoDB DEFAULT CHARSET=utf8;

sqoop export --connect jdbc:mysql://localhost:3306/hive_to_mysql --username root --password sxdx2018 --table user_action --export-dir '/tmp/hive/hdfs/e6429a88-b953-4075-8c96-ad532ea08164/_tmp_space.db/31370dcc-5587-4531-898d-62cbad947f0e' --fields-terminated-by '\t';


檢視執行計劃

我們看下執行計劃

EXPLAIN [EXTENDED|DEPENDENCY|AUTHORIZATION|LOCKS|VECTORIZATION] query

【也可以理解為HQL轉換為mapreduce的過程？？】

官網doc

• EXPLAIN INSERT OVERWRITE TABLE tmp.user_action select * from tmp.bigdata_user;

hive> create temporary table tmp.user_action(id STRING,uid STRING, item_id STRING, behavior_type STRING, item_category STRING, visit_date DATE, province STRING) COMMENT 'temporary table to mysql use sqoop! ' ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' STORED AS TEXTFILE;
OK
Time taken: 1.438 seconds
hive> EXPLAIN INSERT OVERWRITE TABLE tmp.user_action select * from tmp.bigdata_user;
OK
STAGE DEPENDENCIES:
  Stage-1 is a root stage
  Stage-7 depends on stages: Stage-1 , consists of Stage-4, Stage-3, Stage-5
  Stage-4
  Stage-0 depends on stages: Stage-4, Stage-3, Stage-6
  Stage-2 depends on stages: Stage-0
  Stage-3
  Stage-5
  Stage-6 depends on stages: Stage-5

STAGE PLANS:
  Stage: Stage-1
    Map Reduce
      Map Operator Tree: // 發生在job的 map 處理階段
          TableScan          // 讀取表的資料
            alias: bigdata_user // 表的別名
            Statistics: Num rows: 33601 Data size: 15590918 Basic stats: COMPLETE Column stats: NONE
            Select Operator  // 我們select了哪些列
              expressions: id (type: int), uid (type: string), item_id (type: string), behavior_type (type: int), item_category (type: string), visit_date (type: date), province (type: string)
              outputColumnNames: _col0, _col1, _col2, _col3, _col4, _col5, _col6
              Statistics: Num rows: 33601 Data size: 15590918 Basic stats: COMPLETE Column stats: NONE
              File Output Operator
                compressed: false    // 是否有壓縮
                Statistics: Num rows: 33601 Data size: 15590918 Basic stats: COMPLETE Column stats: NONE
                table:
                // 採用了哪些輸入/輸出/壓縮格式
                    input format: org.apache.hadoop.mapred.TextInputFormat
                    output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
                    serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
                    name: tmp.user_action

  Stage: Stage-7
    Conditional Operator  // 條件運算

  Stage: Stage-4
    Move Operator  // 檔案系統相關階段，這裡是去hdfs讀取資料
      files:
          hdfs directory: true
          destination: hdfs://ip10:8020/tmp/hive/hdfs/e6429a88-b953-4075-8c96-ad532ea08164/_tmp_space.db/31370dcc-5587-4531-898d-62cbad947f0e/.hive-staging_hive_2018-07-29_15-24-29_649_851564283197837409-1/-ext-10000

  Stage: Stage-0
    Move Operator
      tables:
          replace: true
          table:  
          // hdfs獲取資料
              input format: org.apache.hadoop.mapred.TextInputFormat
              output format: // 輸出為HiveIgnoreKeyTextOutputFormat格式org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
              serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
              name: tmp.user_action

  Stage: Stage-2
    Stats-Aggr Operator

  Stage: Stage-3
    Map Reduce
      Map Operator Tree:
          TableScan
            File Output Operator
              compressed: false
              table:
                  input format: org.apache.hadoop.mapred.TextInputFormat
                  output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
                  serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
                  name: tmp.user_action

  Stage: Stage-5
    Map Reduce
      Map Operator Tree:
          TableScan
            File Output Operator
              compressed: false
              table:
                  input format: org.apache.hadoop.mapred.TextInputFormat
                  output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
                  serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
                  name: tmp.user_action

  Stage: Stage-6
    Move Operator
      files:
          hdfs directory: true
          destination: hdfs://ip10:8020/tmp/hive/hdfs/e6429a88-b953-4075-8c96-ad532ea08164/_tmp_space.db/31370dcc-5587-4531-898d-62cbad947f0e/.hive-staging_hive_2018-07-29_15-24-29_649_851564283197837409-1/-ext-10000

• 首先列印的是依賴樹，指明那些階段發生在哪些階段之後
  說明：
  Stage-1是最先執行的，Stage-7依賴於Stage-1，Stage-7又由Stage-4、3、5組成

STAGE DEPENDENCIES:
  Stage-1 is a root stage
  Stage-7 depends on stages: Stage-1 , consists of Stage-4, Stage-3, Stage-5
  Stage-4
  Stage-0 depends on stages: Stage-4, Stage-3, Stage-6
  Stage-2 depends on stages: Stage-0
  Stage-3
  Stage-5
  Stage-6 depends on stages: Stage-5

• 接下來可以看出
  stage1、3、5是mapreduce stage
• 還有些檔案系統相關階段、條件運算子、是否壓縮等資訊
• EXPLAIN extended INSERT OVERWRITE TABLE tmp.user_action select * from tmp.bigdata_user;
  這個，列印的資訊太多了，暫時還沒有能看懂的實力。。。

mysql到Hbase

hive、hbase、mysql之間的資料傳輸，以前用過幾次。
1、仍然使用sqoop，換湯不換藥，先不做了。
2、java api讀取hive資料再存入HBase，可參考我以前這篇文章Hbase Java API簡單實踐

使用較大的資料集,為優化做準備

這裡開始,使用較大的資料集進行實驗。

bash ./pre_deal.sh raw_user.csv user_table_big.txt

sudo -u hdfs hdfs dfs -put /home/user_table_big.txt /tmp/data21080729_1/

CREATE EXTERNAL TABLE tmp.optimize(id INT,uid STRING,item_id STRING,behavior_type INT,item_category STRING,visit_date DATE,province STRING) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LOCATION '/tmp/data21080729_1';

• count

  select count(1) from tmp.optimize;
  共20759743條資料，count(1)操作耗時37秒。

• 錯誤的做法，大表join小表

  select * from tmp.optimize a left join tmp.user_action b on(a.id=b.id)limit 5;
  33秒

• 嘗試容易資料傾斜的group by和distinct等

  select distinct a.item_category from optimize a left join bigdata_user b on (a.id = b.id)
  select count(*) from (select count(1) from optimize group by item_category) a;
  資料太小了。各個mapreduce的執行時間差異不大，40秒左右運算結束。
  從事大資料 正式工作前，應該是模擬不出資料傾斜了。
  select count(*) from opyimize where behavior_type=’1’ and visit_date<’2015-12-13’ and visit_date>’2012-12-10’;

偽優化開始

哎，又是理論版

• 合併大量小檔案，以減少map數量
  set mapred.max.split.size
  set mapred.min.split.size.per.node
  set mapred.min.split.size.per.rack等引數

• 控制任務的 reduce 數：

  不指定 reduce 個數的情況下， Hive 會猜測
  確定一個 reduce 個數，基於以下兩個設定：
  引數1：hive.exec.reducers.bytes.per.reducer（每個 reduce 任務處理的資料量，預設為
  1000^3=1G）
  引數2：hive.exec.reducers.max（每個任務最大的 reduce 數，預設為 999）
  計算 reducer 數的公式很簡單 N=min(引數 2，總輸入資料量/引數 1)
  set hive.exec.reducers.bytes.per.reducer
set mapred.reduce.tasks

• 分表
  1、假設原始的資料很大，我們可以搜尋出原始表中的部分資料，然後建立一張表。獲取部分資料我們可以寫python指令碼解析資料。
  2、在Hive Select查詢中一般會掃描整個表內容，會消耗很多時間做沒必要的工作。有時候只需要掃描表中關心的一部分資料，因此建表時引入了partition概念。也就是分目錄的意思。【指定根據哪個欄位進行分割槽就行】
  3、比如說每天的日誌，按日期分表，載入資料到表時，也需要指定partition，然後資料會自動分目錄。

• JVM重用
  每個mapreduce會單獨啟一個JVM。我們可以在配置檔案中設定一些引數，來配置某些task的stage重用JVM，減少任務的開啟時間

• join 時，小表放前面
  hive會將小表進行快取；
  也可以用配置來自動優化：
  set hive.auto.convert.join=true;

• 並行執行
  某些任務之間沒有依賴關係時，就可以設定並行執行
  讓一些不相關的stage能夠並行執行，加速任務的完成

• 考慮表的設計，大表拆分為子表，資料壓縮等
  考慮設定中間和reduce結果進行壓縮，減少佔用空間。

• 資料傾斜

  發生資料傾斜的可能原因：
  key分佈不均勻。
  map端資料傾斜，輸入檔案太多且大小不一 。
  reduce端資料傾斜，分割槽器問題。

使用Hive對資料做一些型別統計的時候遇到某種型別的資料量特別多，而其他型別資料的資料量特別少。當按照型別進行group by的時候，會將相同的group by欄位的reduce任務需要的資料拉取到同一個節點進行聚合，而當其中每一組的資料量過大時，會出現其他組的計算已經完成而這裡還沒計算完成，其他節點的一直等待這個節點的任務執行完成，所以會看到一直map 100% reduce 99%的情況。

　　解決方法：set hive.map.aggr=true
　　　　　　　set hive.groupby.skewindata=true
　　原理：hive.map.aggr=true 這個配置項代表是否在map端進行聚合
　　　　　hive.groupby.skwindata=true 當選項設定為 true，生成的查詢計劃會有兩個 MR Job。第一個 MR Job 中，Map 的輸出結果集合會隨機分佈到 Reduce 中，每個 Reduce 做部分聚合操作，並輸出結果，這樣處理的結果是相同的 Group By Key 有可能被分發到不同的 Reduce 中，從而達到負載均衡的目的；第二個job再進行真正的group by key處理。
　　優化map/reduce數量
　　join小表放進記憶體
　　遇到需要進行join的但是關聯欄位有資料為空，給空值分配隨機的key值。

參考