一、RAID技術(shù)

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

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

[[239212]]

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

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

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

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

RAID0

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

RAID1

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

RAID10

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

RAID3

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

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

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

RAID5

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

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

RAID6

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

2、HDFS應(yīng)用

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

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

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

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

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

三、MapReduce

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

1、MapReduce編程模型

MapReduce是一種非常簡(jiǎn)單又非常強(qiáng)大的編程模型。

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

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

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

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

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); 
    } 
  }} 

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

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

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

map函數(shù)的計(jì)算過(guò)程就是，將這行文本中的單詞提取出來(lái)，針對(duì)每個(gè)單詞輸出一個(gè)這樣的對(duì)。

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

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

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

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

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

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

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

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

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

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

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

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

MapReduce作業(yè)啟動(dòng)和運(yùn)行機(jī)制

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

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

如下圖所示： 

具體作業(yè)啟動(dòng)和計(jì)算過(guò)程如下：

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

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

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

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

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

MapReduce計(jì)算框架處理數(shù)據(jù)合并與連接的操作就在map輸出與reduce輸入之間，這個(gè)過(guò)程有個(gè)專(zhuān)門(mén)的詞匯來(lái)描述，叫做shuffle。 

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

MapReduce框架缺省的Partitioner用key的哈希值對(duì)reduce任務(wù)數(shù)量取模，相同的key一定會(huì)落在相同的reduce任務(wù)id上，實(shí)現(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ù)計(jì)算過(guò)程中發(fā)生奇跡的地方，不管是MapReduce還是Spark，只要是大數(shù)據(jù)批處理計(jì)算，一定會(huì)有shuffle過(guò)程，讓數(shù)據(jù)關(guān)聯(lián)起來(lái)，數(shù)據(jù)的內(nèi)在關(guān)系和價(jià)值才會(huì)呈現(xiàn)出來(lái)。不理解shuffle，就會(huì)在map和reduce編程中產(chǎn)生困惑，不知道該如何正確設(shè)計(jì)map的輸出和reduce的輸入。shuffle也是整個(gè)MapReduce過(guò)程中最難最消耗性能的地方，在MapReduce早期代碼中，一半代碼都是關(guān)于shuffle處理的。

3、工具——Hive

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

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

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

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

首先，看下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)過(guò)shuffle以后，相同的key及其對(duì)應(yīng)的value被放在一起組成一個(gè)，作為輸入交給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>。

計(jì)算過(guò)程如下圖示例： 

這樣一條很有實(shí)用價(jià)值的SQL就這樣被很簡(jiǎn)單的MapReduce計(jì)算過(guò)程處理好了。在數(shù)據(jù)倉(cāng)庫(kù)中，SQL是最常用的分析工具，那么有沒(méi)有能夠自動(dòng)將SQL生成MapReduce代碼的工具呢?這個(gè)工具就是Hadoop大數(shù)據(jù)倉(cāng)庫(kù)Hive。

自動(dòng)將SQL生成MapReduce代碼的工具——Hive

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

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

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

對(duì)于一個(gè)較簡(jiǎn)單的SQL命令，比如：

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

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

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

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

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

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

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

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

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

作者介紹

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