從 MapReduce 到 Hive —— 一次遷移過程小記

@Description(name = “freq
ips”, value = “find most frequence ips from all login ip”, extended = “”)
public class FindFreqIps extends UDF {
    public String evaluate(String content, int limit) {
       // 計算最常用ip的代碼邏輯,并返回結果
       Return result;
    }
}

Add jar hdfs_path/myjar.jar;
Create temporary function FindFreqIps as ‘FindFreqIps’
Select user_id, FindFreqIps(ips) as freqIps from tablexxx

第2步 列轉行,進行join

Select user_id,ip from tablexxx
Lateral view explode(split(ips, “,”))
subview as ip

Select a.user_id as fromid, b.user_id as
toid
SingleIP a
Join  SingleIP b
On a.ip = b.ip and a.user_id <>
b.user_id;

Select * from
(Select a.user_id as fromid, b.user_id as
toid
SingleIP a  Join
SingleIP b
On a.ip = b.ip) m
Where m.fromid <> m.toid;

第一部分是ABSTRACT SYNTAX

第二部分是STAGE DEPENDENCIES

Stage-1 is a root
stage

Stage-0 depends on stages: Stage-1
Stage-3 depends on
stages: Stage-0</pre>這里的stage-1是root stage。而0依賴于1，3依賴于0。 

第三個部分是STAGE PLANS， 就是每個stage中的具體執行的步驟。 
在stage plans里面每一個stage和普通的hadoop程序一樣會有map和reduce的過程。我們截取一段map過程的計劃看下。 

Stage: Stage-1
    Map Reduce
      Alias ->
Map Operator Tree:
        a
          TableScan
            alias:
a
            Reduce
Output Operator
              key expressions:
expr: ip
type: string


         sort


order: +
Map-reduce partition columns:
expr: ip
type: string
              tag:
0
              value expressions:
expr: user_id
type: string</pre>這里是對a表也就是SingleIP表的一個map階段的操作。Reduce output operator這里會顯示使用ip作為key，自增排序，因為是string的所以是字典序的自增。Partition使用ip作為分發字段。tag指的是類似一個來源的概念，因為這里的join采用的是reduce join的方式，每一個從不同的map來的數據最后在reduce進行匯合，他們會被打上一個標記，代表他們的來源。然后就是value的內容，user_id。 



然后再來看看reduce過程的計劃 

Reduce Operator Tree:


    Join

Operator

      condition

map:
Inner Join 0 to 1

      condition

expressions:

        0

{VALUE._col0}

        1

{VALUE._col0}
handleSkewJoin: false
outputColumnNames: _col0, _col2</pre>這里顯示一個join的操作。這里表示把0的內容加到1上。后面有一個handleSkewJoin，這個是hive的一個應對數據傾斜的一種處理方式，默認是關閉的，我們后面再來詳細看。 

這里也可以用explain extended，輸出的信息會更加詳細。那么看了這個我們再比較一下我們之前的第二個查詢計劃，我們來看看加上了嵌套查詢以后的執行計劃有什么變化呢？會發現hive在reduce的執行計劃里面會加上一段 

Filter
Operator
predicate:
expr: (_col0 <> _col2)
type: boolean
在reduce最后輸出之前，進行了一個過濾的操作，過濾的條件就是外部的查詢的where條件。正如我們所料，hive發現這個過程是可以一次性完成的，所以進行了優化，放在了reduce階段來作了。 

另外如果hive中有多張表進行join，如果他們的join key是一樣的，那么hive就會把他們都放在一次mr中完成。 

第3步 數據傾斜 
上一步中，我們計算出了所有的相同ip的人的點對點關系。但是這個結果集會有不少問題，比如如果某個ip是一個公共出口，那么就會出現同一個ip有上萬人都在使用，他們互相join展開以后，結果的數據量會非常大，時間上很慢不說，最終得到的數據實際上很多我們也用不上(這個是基于業務上得考慮)，甚至有可能，在展開的時候會出現各種問題，導致計算時間過長，算不出來。這種情況，我們在hive里面稱之為數據傾斜。 



在group by的時候，如果出現某一個reduce上得數據量過大的情況，hive有一個默認的hive.groupby.skewindata選項，當把它設置為true的時候，hive會將原來的一次MR變成2次，第一次，數據在reduce的時候會隨機分發到每個reduce，做部分的聚合，然后第二次的時候再按照group by的key進行分發。這樣可以有效的處理一般的傾斜情況。 



而在join的時候，如果join的其中某個key的值非常的多，也會導致傾斜。有的時候，如果有null值，在hive看來null和null是相等的，它也會對他們進行join，也會錯誤的傾斜。由于join的時候，hive會把第一張表的內容放到一個內容map中，然后不斷的讀取后表的內容來進行join，所以如果左邊的表示小表這個過程就會非常的高效。當然使用mapjoin也一種有效的方式，直接把一張足夠小的表完全放到內存來后另一張表進行join。類似這樣 

SELECT /*+ MAPJOIN(b) */
a.key, a.value FROM a join b on a.key = b.key；
我們的ip計算使用的是自己join自己，所有也沒有大小表之分，同時單表的數據量也大到無法完全放進內存，那么是不是就要進行硬算呢？在實際中，因為ip的分布沒有傾斜到太過火的程度，硬算也確實可以，但是這里我們換一種方式來稍稍優化一下。 

首先我們采用bucket的方式來保存之間的用戶->ip的數據。使用ip來作為分桶鍵。 

CREATE TABLE userip(user_id bigint, ip STRING)
CLUSTERED BY (ip) INTO 128 BUCKETS;
然后set hive.enforce.bucketing  = true;開啟bucket計算 

from tableaaa 

insert overwrite table tablebbb 

select user_id, ip; 

結果將會被保存到128個不同的桶中，默認根據ip的hashcode來取模。這樣每個桶內的數據基本大概是原數據量的1/100。當然如果原始數據量太大，還可以分桶更加多一些。 

這個地方如果我們不開啟enforce.bucketing的話，也可以通過設置 

set mapred.reduce.tasks=128.然后在查詢中cluster by來強制指定進行分桶。這步完成之后，我們再來進行設置 

set hive.optimize.bucketmapjoin=true; 

set  hive.optimize.bucketmapjoin.sortedmerge=true; 

然后hive就能對每個分塊數據進行mapjoin。 

第4步 用udaf取top N 
好了，現在我們已經有所有的user->user的數據，我們希望要一個user->[users]的一對多的記錄，但是這個數據量有點大，實際上每個用戶大概關聯1000個已經足夠了。首先對數據進行排序，排序的依據就是按照用戶的相同的ip的數量。然后去最前面的1000個，不足的按實際數量取。 

這個地方比較容易想到的就是，先group by fromid，toid，然后count一個總數作為新字段，如下 

這里想到一種做法是用淘寶的一個類sql的row_number實現，然后用row_number來對fromid做主鍵，給按照count從大到小寫上序列編號seq。最后做一個嵌套查詢，只取seq<=1000的數據。Row_number的話標準的hive中沒有。那么這里就可以讓自定義udaf上場了。 

Udaf顧名思義就是一個Aggregate的udf，和之前的udf的區別就是他一般是用來group by的場合中。 

@Description(name = “myudaf”, value = “calc users has most same ips ” )

public class GenericUDAFCollect extends AbstractGenericUDAFResolver {
    @Override
    public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters)
           throws SemanticException {
       return new MyUDAFEvaluator();
    }
}</pre> 
自己定義一個evaluator，并且實現其中的一些方法。 

public static class MyUDAFEvaluator extends GenericUDAFEvaluator {
    @Override
    public ObjectInspector init(Mode m, ObjectInspector[] parameters) throws HiveException {


}
@Override
public void reset(AggregationBuffer agg) throws HiveException {

}
@Override
public AggregationBuffer getNewAggregationBuffer() throws HiveException {

}
// Mapside
@Override
public void iterate(AggregationBuffer agg, Object[] parameters) throws HiveException {

}
// Mapside
@Override
public Object terminatePartial(AggregationBuffer agg) throws HiveException {

}
@Override
public void merge(AggregationBuffer agg, Object partial) throws HiveException {

}
// Reduceside
@Override
public Object terminate(AggregationBuffer agg) throws HiveException {

}

}</pre>在init階段會傳一個Mode進來，這個Mode中定義了以下的幾個階段 

PARTIAL1： 這個是map階段，這個階段會調用iterate(),和terminatePartial() 

PARTIAL2：  這個是map段得combiner階段，會將map端的數據進行合并，也可能沒有這個階段。會執行merge()和terminatePartial() 
FINAL： 這個是reduce階段，會調用merge()和terminate() 
COMPLETE: 這是純map處理，無reduce的情況出現的階段，它會調用iterate()和terminate() 

而從函數方面來說，init是初始化，他會傳入mode作為參數。可以根據不同的階段采取不同的處理。getNewAggregationBuffer的處理是hive為了內存的復用，減少gc，他并不是每一次處理一條記錄都會新申請空間，而是在處理一批數據的時候重復使用一批內存。Terminate就是最終的輸出了。 

Ok，了解了udaf，那么可以動手了。Sql如下 

Select fromid, getTopN(toid,n) from tablexx3
Group by fromid
其中的getTopN首先在map端，將每一個fromid的關聯的toid的次數都記錄下來,記錄條數代表重復的ip數量，然后按照這個次數進行倒序排序，截取前n個。 

在reduce端，將各個map端的結果再按照次數倒序排序，再進行截取n個并進行合并。最終輸出的就是每個fromid對應的toid的列表了。 

從這次從mr轉換到hive的過程中，對我們目前的mr和hive進行了一些比較 

3、mr和hive比較 

1. 運算資源消耗 
無論從時間，數據量，計算量上來看，一般情況下mr都是優于或者等于hive的。mr的靈活性是毋庸置疑的。在轉換到hive的過程中，會有一些為了實現某些場景的需求而不得不用多步hive來實現的時候。 

2. 開發成本/維護成本 
毫無疑問，hive的開發成本是遠低于mr的。如果能熟練的運用udf和transform會更加提高hvie開發的效率。另外對于數據的操作也非常的直觀，對于全世界程序員都喜聞樂見的sql語法的繼承也讓它更加的容易上手。 

   hive獨有的分區管理，方便進行數據的管理。 

   代碼的管理也很方便，就是直接的文本。</span> 

   邏輯的修改和生效很方便。</span> 

   但是當出現異常錯誤的時候，hive的調試會比較麻煩。特別是在大的生產集群上面的時候。</span> 

3. 底層相關性 
在使用hive以后，讀取文件的時候，再也不用關心文件的格式，文件的分隔符，只要指定一次，hive就會保存好。相比mr來說方便了很多。 

當側重關心與業務相關的內容的時候，用hive會比較有優勢。而在一些性能要求高，算法研究的時候，mr會更加適合。 



原文地址：http://rdc.taobao.org/?p=1457