一、RAID技術(shù)

大數(shù)據(jù)技術(shù)主要針對的是大規(guī)模數(shù)據(jù)的計算處理問題，那么要想解決的這一問題，首先要解決的就是大規(guī)模數(shù)據(jù)的存儲問題。大規(guī)模數(shù)據(jù)存儲要解決的核心問題有三個方面：

• 數(shù)據(jù)存儲容量的問題，既然大數(shù)據(jù)要解決的是數(shù)以PB計的數(shù)據(jù)計算問題，而一般的服務(wù)器磁盤容量通常1-2TB，那么如何存儲這么大規(guī)模的數(shù)據(jù)?
• 數(shù)據(jù)讀寫速度的問題，一般磁盤的連續(xù)讀寫速度為幾十MB，以這樣的速度，幾十PB的數(shù)據(jù)恐怕要讀寫到天荒地老……
• 數(shù)據(jù)可靠性的問題，磁盤大約是計算機設(shè)備中最易損壞的硬件了，在網(wǎng)站一塊磁盤使用壽命大概是一年，如果磁盤損壞了，數(shù)據(jù)怎么辦?

[[239212]]

在大數(shù)據(jù)技術(shù)出現(xiàn)之前，人們就需要面對這些關(guān)于存儲的問題，對應(yīng)的解決方案就是RAID技術(shù)。

RAID(獨立磁盤冗余陣列)技術(shù)主要是為了改善磁盤的存儲容量、讀寫速度，增強磁盤的可用性和容錯能力。目前服務(wù)器級別的計算機都支持插入多塊磁盤(8塊或者更多)，通過使用RAID技術(shù)，實現(xiàn)數(shù)據(jù)在多塊磁盤上的并發(fā)讀寫和數(shù)據(jù)備份。

常用RAID技術(shù)有以下幾種，如圖所示： 

常用RAID技術(shù)原理圖

假設(shè)服務(wù)器有N塊磁盤：

RAID0

數(shù)據(jù)在從內(nèi)存緩沖區(qū)寫入磁盤時，根據(jù)磁盤數(shù)量將數(shù)據(jù)分成N份，這些數(shù)據(jù)同時并發(fā)寫入N塊磁盤，使得數(shù)據(jù)整體寫入速度是一塊磁盤的N倍，讀取的時候也一樣，因此RAID0具有極快的數(shù)據(jù)讀寫速度。但是RAID0不做數(shù)據(jù)備份，N塊磁盤中只要有一塊損壞，數(shù)據(jù)完整性就被破壞，所有磁盤的數(shù)據(jù)都會損壞。

RAID1

數(shù)據(jù)在寫入磁盤時，將一份數(shù)據(jù)同時寫入兩塊磁盤，這樣任何一塊磁盤損壞都不會導(dǎo)致數(shù)據(jù)丟失，插入一塊新磁盤就可以通過復(fù)制數(shù)據(jù)的方式自動修復(fù)，具有極高的可靠性。

RAID10

結(jié)合RAID0和RAID1兩種方案，將所有磁盤平均分成兩份，數(shù)據(jù)同時在兩份磁盤寫入，相當(dāng)于RAID1，但是在每一份磁盤里面的N/2塊磁盤上，利用RAID0技術(shù)并發(fā)讀寫，既提高可靠性又改善性能，不過RAID10的磁盤利用率較低，有一半的磁盤用來寫備份數(shù)據(jù)。

RAID3

一般情況下，一臺服務(wù)器上不會出現(xiàn)同時損壞兩塊磁盤的情況，在只損壞一塊磁盤的情況下，如果能利用其它磁盤的數(shù)據(jù)恢復(fù)損壞磁盤的數(shù)據(jù)，就能在保證可靠性和性能的同時，大幅提升磁盤利用率。

在數(shù)據(jù)寫入磁盤的時候，將數(shù)據(jù)分成N-1份，并發(fā)寫入N-1塊磁盤，并在第N塊磁盤記錄校驗數(shù)據(jù)，任何一塊磁盤損壞(包括校驗數(shù)據(jù)磁盤)，都可以利用其它N-1塊磁盤的數(shù)據(jù)修復(fù)。

但是在數(shù)據(jù)修改較多的場景中，任何磁盤修改數(shù)據(jù)都會導(dǎo)致第N塊磁盤重寫校驗數(shù)據(jù)，頻繁寫入的后果是第N塊磁盤比其它磁盤容易損壞，需要頻繁更換，所以RAID3很少在實踐中使用。

RAID5

相比RAID3，更多被使用的方案是RAID5。

RAID5和RAID3很相似，但是校驗數(shù)據(jù)不是寫入第N塊磁盤，而是螺旋式地寫入所有磁盤中。這樣校驗數(shù)據(jù)的修改也被平均到所有磁盤上，避免RAID3頻繁寫壞一塊磁盤的情況。

RAID6

如果數(shù)據(jù)需要很高的可靠性，在出現(xiàn)同時損壞兩塊磁盤的情況下(或者運維管理水平比較落后，壞了一塊磁盤但是遲遲沒有更換，導(dǎo)致又壞了一塊磁盤)，仍然需要修復(fù)數(shù)據(jù)，這時候可以使用RAID6。

RAID6和RAID5類似，但是數(shù)據(jù)只寫入N-2塊磁盤，并螺旋式地在兩塊磁盤中寫入校驗信息(使用不同算法生成)。

在相同磁盤數(shù)目(N)的情況下，各種RAID技術(shù)的比較如下表所示： 

幾種RAID技術(shù)比較

RAID技術(shù)有硬件實現(xiàn)，比如專用的RAID卡或者主板直接支持，也可以通過軟件實現(xiàn)，在操作系統(tǒng)層面將多塊磁盤組成RAID，在邏輯視作一個訪問目錄。RAID技術(shù)在傳統(tǒng)關(guān)系數(shù)據(jù)庫及文件系統(tǒng)中應(yīng)用比較廣泛，是改善計算機存儲特性的重要手段。

RAID技術(shù)只是在單臺服務(wù)器的多塊磁盤上組成陣列，大數(shù)據(jù)需要更大規(guī)模的存儲空間和訪問速度。將RAID技術(shù)原理應(yīng)用到分布式服務(wù)器集群上，就形成了Hadoop分布式文件系統(tǒng)HDFS的架構(gòu)思想。

二、HDFS架構(gòu)思想

1、HDFS架構(gòu)原理

和RAID在多個磁盤上進(jìn)行文件存儲及并行讀寫一樣思路，HDFS在一個大規(guī)模分布式服務(wù)器集群上，對數(shù)據(jù)進(jìn)行并行讀寫及冗余存儲。因為HDFS可以部署在一個比較大的服務(wù)器集群上，集群中所有服務(wù)器的磁盤都可以供HDFS使用，所以整個HDFS的存儲空間可以達(dá)到PB級容量。HDFS架構(gòu)如圖： 

HDFS架構(gòu)

HDFS中關(guān)鍵組件有兩個，一個是NameNode，一個是DataNode。

DataNode負(fù)責(zé)文件數(shù)據(jù)的存儲和讀寫操作，HDFS將文件數(shù)據(jù)分割成若干塊(block)，每個DataNode存儲一部分block，這樣文件就分布存儲在整個HDFS服務(wù)器集群中。應(yīng)用程序客戶端(Client)可以并行對這些數(shù)據(jù)塊進(jìn)行訪問，從而使得HDFS可以在服務(wù)器集群規(guī)模上實現(xiàn)數(shù)據(jù)并行訪問，極大地提高訪問速度。實踐中HDFS集群的DataNode服務(wù)器會有很多臺，一般在幾百臺到幾千臺這樣的規(guī)模，每臺服務(wù)器配有數(shù)塊磁盤，整個集群的存儲容量大概在幾PB到數(shù)百PB。

NameNode負(fù)責(zé)整個分布式文件系統(tǒng)的元數(shù)據(jù)(MetaData)管理，也就是文件路徑名，數(shù)據(jù)block的ID以及存儲位置等信息，承擔(dān)著操作系統(tǒng)中文件分配表(FAT)的角色。HDFS為了保證數(shù)據(jù)的高可用，會將一個block復(fù)制為多份(缺省情況為3份)，并將三份相同的block存儲在不同的服務(wù)器上。這樣當(dāng)有磁盤損壞或者某個DataNode服務(wù)器宕機導(dǎo)致其存儲的block不能訪問的時候，Client會查找其備份的block進(jìn)行訪問。

block多份復(fù)制存儲如下圖所示： 

HDFS的block復(fù)制備份策略 

對于文件/users/sameerp/data/part-0，其復(fù)制備份數(shù)設(shè)置為2，存儲的block ID為1,3，block1的兩個備份存儲在DataNode0和DataNode2兩個服務(wù)器上，block3的兩個備份存儲DataNode4和DataNode6兩個服務(wù)器上，上述任何一臺服務(wù)器宕機后，每個block都至少還有一個備份存在，不會影響對文件/users/sameerp/data/part-0的訪問。

事實上，DataNode會通過心跳和NameNode保持通信，如果DataNode超時未發(fā)送心跳，NameNode就會認(rèn)為這個DataNode已經(jīng)失效，立即查找這個DataNode上存儲的block有哪些，以及這些block還存儲在哪些服務(wù)器上，隨后通知這些服務(wù)器再復(fù)制一份block到其它服務(wù)器上，保證HDFS存儲的block備份數(shù)符合用戶設(shè)置的數(shù)目，即使再有服務(wù)器宕機，也不會丟失數(shù)據(jù)。

2、HDFS應(yīng)用

Hadoop分布式文件系統(tǒng)可以像一般的文件系統(tǒng)那樣進(jìn)行訪問：使用命令行或者編程語言API進(jìn)行文件讀寫操作。我們以HDFS寫文件為例看HDFS處理過程，如下圖： 

HDFS寫文件操作

 

• 應(yīng)用程序Client調(diào)用HDFS API，請求創(chuàng)建文件，HDFS API包含在Client進(jìn)程中;
• HDFS API將請求參數(shù)發(fā)送給NameNode服務(wù)器，NameNode在meta信息中創(chuàng)建文件路徑，并查找DataNode中空閑的block，然后將空閑block的id、對應(yīng)的DataNode服務(wù)器信息返回給Client。因為數(shù)據(jù)塊需要多個備份，所以即使Client只需要一個block的數(shù)據(jù)量，NameNode也會返回多個NameNode信息;
• Client調(diào)用HDFS API，請求將數(shù)據(jù)流寫出;
• HDFS API連接***個DataNode服務(wù)器，將Client數(shù)據(jù)流發(fā)送給DataNode，該DataNode一邊將數(shù)據(jù)寫入本地磁盤，一邊發(fā)送給第二個DataNode，同理第二個DataNode記錄數(shù)據(jù)并發(fā)送給第三個DataNode;
• Client通知NameNode文件寫入完成，NameNode將文件標(biāo)記為正常，可以進(jìn)行讀操作了。

HDFS雖然提供了API，但是在實踐中，我們很少自己編程直接去讀取HDFS中的數(shù)據(jù)，原因正如開篇提到，在大數(shù)據(jù)場景下，移動計算比移動數(shù)據(jù)更劃算。

與其寫程序去讀取分布在這么多DataNode上的數(shù)據(jù)，不如將程序分發(fā)到DataNode上去訪問其上的block數(shù)據(jù)。但是如何對程序進(jìn)行分發(fā)?分發(fā)出去的程序又如何訪問HDFS上的數(shù)據(jù)?計算的結(jié)果如何處理，如果結(jié)果需要合并，該如何合并?

Hadoop提供了對存儲在HDFS上的大規(guī)模數(shù)據(jù)進(jìn)行并行計算的框架，就是MapReduce。

三、MapReduce

Hadoop解決大規(guī)模數(shù)據(jù)分布式計算的方案是MapReduce。MapReduce既是一個編程模型，又是一個計算框架。也就是說，開發(fā)人員必須基于MapReduce編程模型進(jìn)行編程開發(fā)，然后將程序通過MapReduce計算框架分發(fā)到Hadoop集群中運行。我們先看一下作為編程模型的MapReduce。

1、MapReduce編程模型

MapReduce是一種非常簡單又非常強大的編程模型。

簡單在于其編程模型只包含map和reduce兩個過程，map的主要輸入是一對值，經(jīng)過map計算后輸出一對值;然后將相同key合并，形成;再將這個輸入reduce，經(jīng)過計算輸出零個或多個對。

但是MapReduce同時又是非常強大的，不管是關(guān)系代數(shù)運算(SQL計算)，還是矩陣運算(圖計算)，大數(shù)據(jù)領(lǐng)域幾乎所有的計算需求都可以通過MapReduce編程來實現(xiàn)。

我們以WordCount程序為例。WordCount主要解決文本處理中的詞頻統(tǒng)計問題，就是統(tǒng)計文本中每一個單詞出現(xiàn)的次數(shù)。如果只是統(tǒng)計一篇文章的詞頻，幾十K到幾M的數(shù)據(jù)，那么寫一個程序，將數(shù)據(jù)讀入內(nèi)存，建一個Hash表記錄每個詞出現(xiàn)的次數(shù)就可以了，如下圖： 

小數(shù)據(jù)量的詞頻統(tǒng)計

 

但是如果想統(tǒng)計全世界互聯(lián)網(wǎng)所有網(wǎng)頁(數(shù)萬億計)的詞頻數(shù)(這正是google這樣的搜索引擎典型需求)，你不可能寫一個程序把全世界的網(wǎng)頁都讀入內(nèi)存，這時候就需要用MapReduce編程來解決。

public class WordCount { 
  public static class TokenizerMapper 
       extends Mapper<Object, Text, Text, IntWritable>{ 
    private final static IntWritable one = new IntWritable(1); 
    private Text word = new Text(); 
    public void map(Object key, Text value, Context context 
                    ) throws IOException, InterruptedException { 
      StringTokenizer itr = new StringTokenizer(value.toString()); 
      while (itr.hasMoreTokens()) { 
        word.set(itr.nextToken()); 
        context.write(word, one); 
      } 
    } 
  } 
  public static class IntSumReducer 
       extends Reducer<Text,IntWritable,Text,IntWritable> { 
    private IntWritable result = new IntWritable(); 
    public void reduce(Text key, Iterable<IntWritable> values, 
                       Context context 
                       ) throws IOException, InterruptedException { 
      int sum = 0; 
      for (IntWritable val : values) { 
        sum += val.get(); 
      } 
      result.set(sum); 
      context.write(key, result); 
    } 
  }} 

其核心是一個map函數(shù)，一個reduce函數(shù)。

map函數(shù)的輸入主要是一個對，在這個例子里，value是要統(tǒng)計的所有文本中的一行數(shù)據(jù)，key在這里不重要，我們忽略。

public void map(Object key, Text value, Context context 
                    ) 

map函數(shù)的計算過程就是，將這行文本中的單詞提取出來，針對每個單詞輸出一個這樣的對。

MapReduce計算框架會將這些收集起來，將相同的word放在一起，形成>這樣的數(shù)據(jù)，然后將其輸入給reduce函數(shù)。

public void reduce(Text key, Iterable<IntWritable> values, 
                      Context context 
                      ) 

這里的reduce的輸入?yún)?shù)values就是由很多個1組成的集合，而key就是具體的單詞word。

reduce函數(shù)的計算過程就是，將這個集合里的1求和，再將單詞(word)和這個和(sum)組成一個()輸出。每一個輸出就是一個單詞和它的詞頻統(tǒng)計總和。

假設(shè)有兩個block的文本數(shù)據(jù)需要進(jìn)行詞頻統(tǒng)計，MapReduce計算過程如下圖： 

MapReduce計算過程

 

一個map函數(shù)可以針對一部分?jǐn)?shù)據(jù)進(jìn)行運算，這樣就可以將一個大數(shù)據(jù)切分成很多塊(這也正是HDFS所做的)，MapReduce計算框架為每個塊分配一個map函數(shù)去計算，從而實現(xiàn)大數(shù)據(jù)的分布式計算。

2、MapReduce計算框架架構(gòu)原理

前面提到MapReduce編程模型將大數(shù)據(jù)計算過程切分為map和reduce兩個階段，在map階段為每個數(shù)據(jù)塊分配一個map計算任務(wù)，然后將所有map輸出的key進(jìn)行合并，相同的key及其對應(yīng)的value發(fā)送給同一個reduce任務(wù)去處理。

這個過程有兩個關(guān)鍵問題需要處理：

1. 如何為每個數(shù)據(jù)塊分配一個map計算任務(wù)，代碼是如何發(fā)送數(shù)據(jù)塊所在服務(wù)器的，發(fā)送過去是如何啟動的，啟動以后又如何知道自己需要計算的數(shù)據(jù)在文件什么位置(數(shù)據(jù)塊id是什么)?
2. 處于不同服務(wù)器的map輸出的 ，如何把相同的key聚合在一起發(fā)送給reduce任務(wù)?

這兩個關(guān)鍵問題正好對應(yīng)前面文章中“MapReduce計算過程”一圖中兩處“MapReduce框架處理”： 

MapReduce計算過程中兩處MapReduce框架處理

 

我們先看下MapReduce是如何啟動處理一個大數(shù)據(jù)計算應(yīng)用作業(yè)的：

MapReduce作業(yè)啟動和運行機制

我們以Hadoop1為例，MapReduce運行過程涉及以下幾類關(guān)鍵進(jìn)程：

• 大數(shù)據(jù)應(yīng)用進(jìn)程：啟動用戶MapReduce程序的主入口，主要指定Map和Reduce類、輸入輸出文件路徑等，并提交作業(yè)給Hadoop集群。
• JobTracker進(jìn)程：根據(jù)要處理的輸入數(shù)據(jù)量啟動相應(yīng)數(shù)量的map和reduce進(jìn)程任務(wù)，并管理整個作業(yè)生命周期的任務(wù)調(diào)度和監(jiān)控。JobTracker進(jìn)程在整個Hadoop集群全局唯一。
• TaskTracker進(jìn)程：負(fù)責(zé)啟動和管理map進(jìn)程以及reduce進(jìn)程。因為需要每個數(shù)據(jù)塊都有對應(yīng)的map函數(shù)，TaskTracker進(jìn)程通常和HDFS的DataNode進(jìn)程啟動在同一個服務(wù)器，也就是說，Hadoop集群中絕大多數(shù)服務(wù)器同時運行DataNode進(jìn)程和TaskTacker進(jìn)程。

如下圖所示： 

MapReduce作業(yè)啟動和運行機制

 

具體作業(yè)啟動和計算過程如下：

• 應(yīng)用進(jìn)程將用戶作業(yè)jar包存儲在HDFS中，將來這些jar包會分發(fā)給Hadoop集群中的服務(wù)器執(zhí)行MapReduce計算;
• 應(yīng)用程序提交job作業(yè)給JobTracker;
• JobTacker根據(jù)作業(yè)調(diào)度策略創(chuàng)建JobInProcess樹，每個作業(yè)都會有一個自己的JobInProcess樹;
• JobInProcess根據(jù)輸入數(shù)據(jù)分片數(shù)目(通常情況就是數(shù)據(jù)塊的數(shù)目)和設(shè)置的reduce數(shù)目創(chuàng)建相應(yīng)數(shù)量的TaskInProcess;
• TaskTracker進(jìn)程和JobTracker進(jìn)程進(jìn)行定時通信;
• 如果TaskTracker有空閑的計算資源(空閑CPU核)，JobTracker就會給它分配任務(wù)。分配任務(wù)的時候會根據(jù)TaskTracker的服務(wù)器名字匹配在同一臺機器上的數(shù)據(jù)塊計算任務(wù)給它，使啟動的計算任務(wù)正好處理本機上的數(shù)據(jù)，以實現(xiàn)我們一開始就提到的“移動計算比移動數(shù)據(jù)更劃算”;
• TaskRunner收到任務(wù)后根據(jù)任務(wù)類型(map還是reduce)，任務(wù)參數(shù)(作業(yè)jar包路徑，輸入數(shù)據(jù)文件路徑，要處理的數(shù)據(jù)在文件中的起始位置和偏移量，數(shù)據(jù)塊多個備份的DataNode主機名等)啟動相應(yīng)的map或者reduce進(jìn)程;
• map或者reduce程序啟動后，檢查本地是否有要執(zhí)行任務(wù)的jar包文件，如果沒有，就去HDFS上下載，然后加載map或者reduce代碼開始執(zhí)行;
• 如果是map進(jìn)程，從HDFS讀取數(shù)據(jù)(通常要讀取的數(shù)據(jù)塊正好存儲在本機);如果是reduce進(jìn)程，將結(jié)果數(shù)據(jù)寫出到HDFS。

通過以上過程，MapReduce可以將大數(shù)據(jù)作業(yè)計算任務(wù)分布在整個Hadoop集群中運行，每個map計算任務(wù)要處理的數(shù)據(jù)通常都能從本地磁盤上讀取到，而用戶要做的僅僅是編寫一個map函數(shù)和一個reduce函數(shù)就可以了，根本不用關(guān)心這兩個函數(shù)是如何被分布啟動到集群上的，數(shù)據(jù)塊又是如何分配給計算任務(wù)的。這一切都由MapReduce計算框架完成。

MapReduce數(shù)據(jù)合并與連接機制

在WordCount例子中，要統(tǒng)計相同單詞在所有輸入數(shù)據(jù)中出現(xiàn)的次數(shù)，而一個map只能處理一部分?jǐn)?shù)據(jù)，一個熱門單詞幾乎會出現(xiàn)在所有的map中，這些單詞必須要合并到一起進(jìn)行統(tǒng)計才能得到正確的結(jié)果。

事實上，幾乎所有的大數(shù)據(jù)計算場景都需要處理數(shù)據(jù)關(guān)聯(lián)的問題，簡單如WordCount只要對key進(jìn)行合并就可以了，復(fù)雜如數(shù)據(jù)庫的join操作，需要對兩種類型(或者更多類型)的數(shù)據(jù)根據(jù)key進(jìn)行連接。

MapReduce計算框架處理數(shù)據(jù)合并與連接的操作就在map輸出與reduce輸入之間，這個過程有個專門的詞匯來描述，叫做shuffle。 

MapReduce shuffle過程

 

每個map任務(wù)的計算結(jié)果都會寫入到本地文件系統(tǒng)，等map任務(wù)快要計算完成的時候，MapReduce計算框架會啟動shuffle過程，在map端調(diào)用一個Partitioner接口，對map產(chǎn)生的每個進(jìn)行reduce分區(qū)選擇，然后通過http通信發(fā)送給對應(yīng)的reduce進(jìn)程。這樣不管map位于哪個服務(wù)器節(jié)點，相同的key一定會被發(fā)送給相同的reduce進(jìn)程。reduce端對收到的進(jìn)行排序和合并，相同的key放在一起，組成一個傳遞給reduce執(zhí)行。

MapReduce框架缺省的Partitioner用key的哈希值對reduce任務(wù)數(shù)量取模，相同的key一定會落在相同的reduce任務(wù)id上，實現(xiàn)上，這樣的Partitioner代碼只需要一行，如下所示：

/** Use {@link Object#hashCode()} to partition. */ 
public int getPartition(K2 key, V2 value, int numReduceTasks) { 
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks; 
 } 

shuffle是大數(shù)據(jù)計算過程中發(fā)生奇跡的地方，不管是MapReduce還是Spark，只要是大數(shù)據(jù)批處理計算，一定會有shuffle過程，讓數(shù)據(jù)關(guān)聯(lián)起來，數(shù)據(jù)的內(nèi)在關(guān)系和價值才會呈現(xiàn)出來。不理解shuffle，就會在map和reduce編程中產(chǎn)生困惑，不知道該如何正確設(shè)計map的輸出和reduce的輸入。shuffle也是整個MapReduce過程中最難最消耗性能的地方，在MapReduce早期代碼中，一半代碼都是關(guān)于shuffle處理的。

3、工具——Hive

既然MapReduce計算模型可以解決絕大多數(shù)的數(shù)據(jù)分析與數(shù)據(jù)挖掘任務(wù)，那么對于如下我們常見的一條SQL分析語句，MapReduce如何編程實現(xiàn)?

SELECT pageid, age, count(1) FROM pv_users GROUP BY pageid, age; 

這是一條非常常見的SQL統(tǒng)計分析語句，統(tǒng)計不同年齡的用戶訪問不同網(wǎng)頁的興趣偏好，對于產(chǎn)品運營和設(shè)計很有價值。具體數(shù)據(jù)輸入和執(zhí)行結(jié)果如下圖示例： 

group by輸入輸出示例

 

左邊是要分析的數(shù)據(jù)表，右邊是分析結(jié)果。實際上把左邊表相同的行累計求和，就得到右邊的表了，看起來跟WordCount的計算很一樣。確實也是這樣，我們看下這條SQL語句的MapReduce的計算過程，map和reduce函數(shù)的輸入輸出以及函數(shù)處理過程分別是什么樣。

首先，看下map函數(shù)的輸入key和value，key不重要，忽略掉，value就是左邊表中每一行的數(shù)據(jù)，<1, 25>這樣。map函數(shù)的輸出就是以輸入的value作為key，value統(tǒng)一設(shè)為1，<<1, 25>, 1>這樣。

map函數(shù)的輸出經(jīng)過shuffle以后，相同的key及其對應(yīng)的value被放在一起組成一個，作為輸入交給reduce函數(shù)處理。如<<2, 25>, 1>被map函數(shù)輸出兩次，那么到了reduce這里，就變成輸入<<2, 25>, <1, 1>>，key是<2, 25>， value集合是<1, 1>。在reduce函數(shù)內(nèi)部，value集合里所有的數(shù)字被相加，然后輸出。reduce的輸出就是<<2, 25>, 2>。

計算過程如下圖示例： 

group by的MapReduce計算過程示例

 

這樣一條很有實用價值的SQL就這樣被很簡單的MapReduce計算過程處理好了。在數(shù)據(jù)倉庫中，SQL是最常用的分析工具，那么有沒有能夠自動將SQL生成MapReduce代碼的工具呢?這個工具就是Hadoop大數(shù)據(jù)倉庫Hive。

自動將SQL生成MapReduce代碼的工具——Hive

Hive能夠直接處理用戶輸入的SQL語句(Hive的SQL語法和數(shù)據(jù)庫標(biāo)準(zhǔn)SQL略有不同)，調(diào)用MapReduce計算框架完成數(shù)據(jù)分析操作。具體架構(gòu)如下圖： 

Hive架構(gòu)

 

用戶通過Hive的Client(Hive的命令行工具，JDBC等)向Hive提交SQL命令。如果是創(chuàng)建數(shù)據(jù)表的DDL語句，Hive就會通過執(zhí)行引擎Driver將數(shù)據(jù)表的信息記錄在Metastore組件中，這個組件通常用一個關(guān)系數(shù)據(jù)庫實現(xiàn)，記錄表名、字段名、字段類型、關(guān)聯(lián)HDFS文件路徑等這些數(shù)據(jù)庫的meta信息(元信息)。

如果用戶提交的是查詢分析數(shù)據(jù)的DQL語句，Driver就會將該語句提交給自己的編譯器Compiler進(jìn)行語法分析、語法解析、語法優(yōu)化等一系列操作，***生成一個MapReduce執(zhí)行計劃。然后根據(jù)該執(zhí)行計劃生成一個MapReduce的作業(yè)，提交給Hadoop MapReduce計算框架處理。

對于一個較簡單的SQL命令，比如：

SELECT * FROM status_updates WHERE status LIKE ‘michael jackson’; 

其對應(yīng)的Hive執(zhí)行計劃如下圖： 

Hive執(zhí)行計劃示例

 

Hive內(nèi)部預(yù)置了很多函數(shù)，Hive的執(zhí)行計劃就是根據(jù)SQL語句生成這些函數(shù)的DAG(有向無環(huán)圖)，然后封裝進(jìn)MapReduce的map和reduce函數(shù)中。這個例子中，map函數(shù)調(diào)用了三個Hive內(nèi)置函數(shù)TableScanOpoerator、FilterOperator、FileOutputOperator，就完成了map計算，而且無需reduce函數(shù)。

除了上面這些簡單的聚合(group by)、過濾(where)操作，Hive還能執(zhí)行連接(join on)操作。上面例子中，pv_users表的數(shù)據(jù)在實際中是無法直接得到的，因為pageid數(shù)據(jù)來自用戶訪問日志，每個用戶進(jìn)行一次頁面瀏覽，就會生成一條訪問記錄，保存在page_view表中。而年齡age信息則記錄在用戶表user中。如下圖： 

page_view表和user表示例

 

這兩張表都有一個相同的字段userid，根據(jù)這個字段可以將兩張表連接起來，生成前面的pv_users表，SQL命令如下：

SELECT pv.pageid, u.age FROM page_view pv JOIN user u ON (pv.userid = u.userid); 

同樣，這個SQL命令也可以轉(zhuǎn)化為MapReduce計算，如下圖： 

join的MapReduce計算過程示例

 

join的MapReduce計算過程和前面的group by稍有不同，因為join涉及兩張表，來自兩個文件(夾)，所以需要在map輸出的時候進(jìn)行標(biāo)記，比如來自***張表的輸出value就記錄為<1, X>，這里的1表示數(shù)據(jù)來自***張表。這樣經(jīng)過shuffle以后，相同的key被輸入到同一個reduce函數(shù)，就可以根據(jù)表的標(biāo)記對value數(shù)據(jù)求笛卡爾積，輸出就join的結(jié)果。

在實踐中，工程師并不需要經(jīng)常編寫MapReduce程序，因為網(wǎng)站最主要的大數(shù)據(jù)處理就是SQL分析，在Facebook，據(jù)說90%以上的MapReduce任務(wù)都是Hive產(chǎn)生的。Hive在大數(shù)據(jù)應(yīng)用中的作用非常重要。

作者介紹

李智慧，《大型網(wǎng)站技術(shù)架構(gòu)：核心原理與案例分析》作者。曾供職于阿里巴巴與英特爾亞太研發(fā)中心，從事大型網(wǎng)站與大數(shù)據(jù)方面的研發(fā)工作，目前在做企業(yè)級區(qū)塊鏈方面的開發(fā)工作。