轉載來自：http://www.codeweblog.com/hive-%E5%9F%BA%E7%A1%80-1-%E5%88%86%E5%8C%BA-%E6%A1%B6-sort-merge-bucket-join/

Hive 已是目前業界最為通用、廉價的構建大資料時代資料倉庫的解決方案了，雖然也有 Impala 等後起之秀，但目前從功能、穩定性等方面來說，Hive 的地位尚不可撼動。

其實這篇博文主要是想聊聊 SMB join 的，Join 是整個 MR/Hive 最為核心的部分之一，是每個 Hadoop/Hive/DW RD 必須掌握的部分，之前也有幾篇文章聊到過 MR/Hive 中的 join，其實底層都是相同的，只是上層做了些封裝而已，如果你還不瞭解究竟 Join 有哪些方式，以及底層怎麼實現的，請參考如下連結：

http://my.codeweblog.com/leejun2005/blog/95186 MapReduce 中的兩表 join 幾種方案簡介

http://my.codeweblog.com/leejun2005/blog/111963 Hadoop 多表 join：map side join 範例

http://my.codeweblog.com/leejun2005/blog/158491 Hive & Performance 學習筆記

在最後一篇連結中，有這麼兩副圖：

前面兩個很好理解，基本上每個人都會接觸到，但最後一種，可能有同學還是比較陌生，SMB 存在的目的主要是為了解決大表與大表間的 Join 問題，分桶其實就是把大表化成了“小表”，然後 Map-Side Join 解決之，這是典型的分而治之的思想。在聊 SMB Join 之前，我們還是先複習下相關的基礎概念。

1、Hive 分割槽表

在Hive Select查詢中一般會掃描整個表內容，會消耗很多時間做沒必要的工作。有時候只需要掃描表中關心的一部分資料，因此建表時引入了partition概念。分割槽表指的是在建立表時指定的partition的分割槽空間。

Hive可以對資料按照某列或者某些列進行分割槽管理，所謂分割槽我們可以拿下面的例子進行解釋。
當前網際網路應用每天都要儲存大量的日誌檔案，幾G、幾十G甚至更大都是有可能。儲存日誌，其中必然有個屬性是日誌產生的日期。在產生分割槽時，就可以按照日誌產生的日期列進行劃分。把每一天的日誌當作一個分割槽。
將資料組織成分割槽，主要可以提高資料的查詢速度。至於使用者儲存的每一條記錄到底放到哪個分割槽，由使用者決定。即使用者在載入資料的時候必須顯示的指定該部分資料放到哪個分割槽。

1.1 實現細節

1、一個表可以擁有一個或者多個分割槽，每個分割槽以資料夾的形式單獨存在表文件夾的目錄下。
2、表和列名不區分大小寫。
3、分割槽是以欄位的形式在表結構中存在，通過describe table命令可以檢視到欄位存在， 但是該欄位不存放實際的資料內容，僅僅是分割槽的表示（偽列） 。

1.2 語法

1. 建立一個分割槽表，以 ds 為分割槽列：
create table invites (id int, name string) partitioned by (ds string) row format delimited fields terminated by 't' stored as textfile;
2. 將資料新增到時間為 2013-08-16 這個分割槽中：
load data local inpath '/home/hadoop/Desktop/data.txt' overwrite into table invites partition (ds='2013-08-16');
3. 將資料新增到時間為 2013-08-20 這個分割槽中：
load data local inpath '/home/hadoop/Desktop/data.txt' overwrite into table invites partition (ds='2013-08-20');
4. 從一個分割槽中查詢資料：
select * from invites where ds ='2013-08-12';
5. 往一個分割槽表的某一個分割槽中新增資料：
insert overwrite table invites partition (ds='2013-08-12') select id,max(name) from test group by id;
可以檢視分割槽的具體情況，使用命令：
hadoop fs -ls /home/hadoop.hive/warehouse/invites
或者：
show partitions tablename;

2、Hive 桶

對於每一個表（table）或者分割槽， Hive可以進一步組織成桶，也就是說桶是更為細粒度的資料範圍劃分。Hive也是 針對某一列進行桶的組織。Hive採用對列值雜湊，然後除以桶的個數求餘的方式決定該條記錄存放在哪個桶當中。

把表（或者分割槽）組織成桶（Bucket）有兩個理由：

（1）獲得更高的查詢處理效率。桶為表加上了額外的結構，Hive 在處理有些查詢時能利用這個結構。具體而言，連線兩個在（包含連線列的）相同列上劃分了桶的表，可以使用 Map 端連線 （Map-side join）高效的實現。比如JOIN操作。對於JOIN操作兩個表有一個相同的列，如果對這兩個表都進行了桶操作。那麼將儲存相同列值的桶進行JOIN操作就可以，可以大大較少JOIN的資料量。

（2）使取樣（sampling）更高效。在處理大規模資料集時，在開發和修改查詢的階段，如果能在資料集的一小部分資料上試執行查詢，會帶來很多方便。

1. 建立帶桶的 table ：

create table bucketed_user(id int,name string) clustered by (id) sorted by(name) into 4 buckets row format delimited fields terminated by '\t' stored as textfile;
首先，我們來看如何告訴Hive—個表應該被劃分成桶。我們使用CLUSTERED BY 子句來指定劃分桶所用的列和要劃分的桶的個數：

CREATE TABLE bucketed_user (id INT) name STRING)
CLUSTERED BY (id) INTO 4 BUCKETS;

在這裡，我們使用使用者ID來確定如何劃分桶(Hive使用對值進行雜湊並將結果除 以桶的個數取餘數。這樣，任何一桶裡都會有一個隨機的使用者集合（PS：其實也能說是隨機，不是嗎？）。

對於map端連線的情況，兩個表以相同方式劃分桶。處理左邊表內某個桶的 mapper知道右邊表內相匹配的行在對應的桶內。因此，mapper只需要獲取那個桶 (這只是右邊表記憶體儲資料的一小部分)即可進行連線。這一優化方法並不一定要求 兩個表必須桶的個數相同，兩個表的桶個數是倍數關係也可以。用HiveQL對兩個劃分了桶的表進行連線，可參見“map連線”部分（P400）。

桶中的資料可以根據一個或多個列另外進行排序。由於這樣對每個桶的連線變成了高效的歸併排序(merge-sort), 因此可以進一步提升map端連線的效率。以下語法宣告一個表使其使用排序桶：

CREATE TABLE bucketed_users (id INT, name STRING)
CLUSTERED BY (id) SORTED BY (id ASC) INTO 4 BUCKETS;

我們如何保證表中的資料都劃分成桶了呢？把在Hive外生成的資料載入到劃分成 桶的表中，當然是可以的。其實讓Hive來劃分桶更容易。這一操作通常針對已有的表。

Hive並不檢查資料檔案中的桶是否和表定義中的桶一致(無論是對於桶 的數量或用於劃分桶的列）。如果兩者不匹配，在査詢時可能會碰到錯 誤或未定義的結果。因此，建議讓Hive來進行劃分桶的操作。

有一個沒有劃分桶的使用者表：
hive> SELECT * FROM users;
0 Nat
2 Doe
B Kay
4 Ann

2. 強制多個 reduce 進行輸出：

要向分桶表中填充成員，需要將 hive.enforce.bucketing 屬性設定為 true。①這 樣，Hive 就知道用表定義中宣告的數量來建立桶。然後使用 INSERT 命令即可。需要注意的是： clustered by和sorted by不會影響資料的匯入，這意味著，使用者必須自己負責資料如何如何匯入，包括資料的分桶和排序。
'set hive.enforce.bucketing = true' 可以自動控制上一輪reduce的數量從而適配bucket的個數，當然，使用者也可以自主設定mapred.reduce.tasks去適配bucket個數，推薦使用'set hive.enforce.bucketing = true'

3. 往表中插入資料：

INSERT OVERWRITE TABLE bucketed_users SELECT * FROM users;

物理上，每個桶就是表(或分割槽）目錄裡的一個檔案。它的檔名並不重要，但是桶 n 是按照字典序排列的第 n 個檔案。事實上，桶對應於 MapReduce 的輸出檔案分割槽：一個作業產生的桶(輸出檔案)和reduce任務個數相同。我們可以通過檢視剛才 建立的bucketd_users表的佈局來了解這一情況。執行如下命令：

4. 查看錶的結構：

hive> dfs -ls /user/hive/warehouse/bucketed_users;
將顯示有4個新建的檔案。檔名如下(檔名包含時間戳，由Hive產生，因此 每次執行都會改變)：
attempt_201005221636_0016_r_000000_0
attempt_201005221636_0016_r-000001_0
attempt_201005221636_0016_r_000002_0
attempt_201005221636_0016_r_000003_0
第一個桶裡包括使用者IDO和4，因為一個INT的雜湊值就是這個整數本身，在這裡 除以桶數(4)以後的餘數：②

5. 讀取資料，看每一個檔案的資料：

hive> dfs -cat /user/hive/warehouse/bucketed_users/*0_0;
0 Nat
4 Ann

用TABLESAMPLE子句對錶進行取樣，我們可以獲得相同的結果。這個子句會將 查詢限定在表的一部分桶內，而不是使用整個表：

6. 對桶中的資料進行取樣：

hive> SELECT * FROM bucketed_users
> TABLESAMPLE(BUCKET 1 OUT OF 4 ON id);
0 Nat
4 Ann

桶的個數從1開始計數。因此，前面的查詢從4個桶的第一個中獲取所有的使用者。 對於一個大規模的、均勻分佈的資料集，這會返回表中約四分之一的資料行。我們 也可以用其他比例對若干個桶進行取樣(因為取樣並不是一個精確的操作，因此這個 比例不一定要是桶數的整數倍)。例如，下面的查詢返回一半的桶：

7. 查詢一半返回的桶數：

hive> SELECT * FROM bucketed_users
> TABLESAMPLE(BUCKET 1 OUT OF 2 ON id)；
0 Nat
4 Ann
2 Joe

因為查詢只需要讀取和TABLESAMPLE子句匹配的桶，所以取樣分桶表是非常高效 的操作。如果使用rand()函式對沒有劃分成桶的表進行取樣，即使只需要讀取很 小一部分樣本，也要掃描整個輸入資料集：

hive〉 SELECT * FROM users
> TABLESAMPLE(BUCKET 1 OUT OF 4 ON rand());
2 Doe

①從Hive 0.6.0開始，對以前的版本，必須把mapred.reduce .tasks設為表中要填 充的桶的個數。如果桶是排序的，還需要把hive.enforce.sorting設為true。
②顯式原始檔案時，因為分隔字元是一個不能列印的控制字元，因此欄位都擠在一起。

3、舉個完整的小栗子：

（1）建student & student1 表：

create table student(id INT, age INT, name STRING)
partitioned by(stat_date STRING)
clustered by(id) sorted by(age) into 2 buckets
row format delimited fields terminated by ',';

create table student1(id INT, age INT, name STRING)
partitioned by(stat_date STRING)
clustered by(id) sorted by(age) into 2 buckets
row format delimited fields terminated by ',';

（2）設定環境變數：

set hive.enforce.bucketing = true;

（3）插入資料：

cat bucket.txt

1,20,zxm
2,21,ljz
3,19,cds
4,18,mac
5,22,android
6,23,symbian
7,25,wp

LOAD DATA local INPATH '/home/lijun/bucket.txt' OVERWRITE INTO TABLE student partition(stat_date="20120802");

from student
insert overwrite table student1 partition(stat_date="20120802")
select id,age,name where stat_date="20120802" sort by age;

（4）檢視檔案目錄：

hadoop fs -ls /hive/warehouse/test.db/student1/stat_date=20120802
Found 2 items
-rw-r--r-- 2 lijun supergroup 31 2013-11-24 19:16 /hive/warehouse/test.db/student1/stat_date=20120802/000000_0
-rw-r--r-- 2 lijun supergroup 39 2013-11-24 19:16 /hive/warehouse/test.db/student1/stat_date=20120802/000001_0

（5）檢視sampling資料：

hive> select * from student1 tablesample(bucket 1 out of 2 on id);

Total MapReduce jobs = 1
Launching Job 1 out of 1
.......
OK
4 18 mac 20120802
2 21 ljz 20120802
6 23 symbian 20120802
Time taken: 20.608 seconds

注：tablesample是抽樣語句，語法：TABLESAMPLE(BUCKET x OUT OF y)
y必須是table總bucket數的倍數或者因子。hive根據y的大小，決定抽樣的比例。例如，table總共分了64份，當y=32時，抽取(64/32=)2個bucket的資料，當y=128時，抽取(64/128=)1/2個bucket的資料。x表示從哪個bucket開始抽取。例如，table總bucket數為32，tablesample(bucket 3 out of 16)，表示總共抽取（32/16=）2個bucket的資料，分別為第3個bucket和第（3+16=）19個bucket的資料。

4、Refer:

hive--Sort Merge Bucket Map Join