大數(shù)據(jù)導(dǎo)論-第2章-大數(shù)據(jù)生態(tài)系統(tǒng)

大數(shù)據(jù)技術(shù)導(dǎo)論第二章大數(shù)據(jù)生態(tài)系統(tǒng)2.1

認(rèn)識(shí)Hadoop2.2

HDFS2.3

MapReduce2.4

zookeeper1.3.2數(shù)據(jù)存儲(chǔ)技術(shù)分布式數(shù)據(jù)庫Bigtable分布式文件系統(tǒng)GFS分布式并行處理技術(shù)MapreduceHDFSHBaseMapReduce2.1認(rèn)識(shí)hadoop第二章大數(shù)據(jù)生態(tài)系統(tǒng)1.1.1大數(shù)據(jù)產(chǎn)生歷史必然——谷歌三駕馬車2.1認(rèn)識(shí)hadoop第二章大數(shù)據(jù)生態(tài)系統(tǒng)第二章大數(shù)據(jù)生態(tài)系統(tǒng)2.2

HDFS2.1認(rèn)識(shí)Hadoop2.3

MapReduce2.4

zookeeper2.2.1HDFS體系結(jié)構(gòu)2.2HDFS第二章大數(shù)據(jù)生態(tài)系統(tǒng)2.2.1HDFS體系結(jié)構(gòu)2.2HDFS第二章大數(shù)據(jù)生態(tài)系統(tǒng)2.2.1HDFS體系結(jié)構(gòu)2.2HDFS第二章大數(shù)據(jù)生態(tài)系統(tǒng)(1)HDFS=1個(gè)NameNode+1個(gè)SecondaryNameNode+多個(gè)DataNode；(2)NameNode稱為名稱節(jié)點(diǎn)、命名空間、主節(jié)點(diǎn)、元數(shù)據(jù)節(jié)點(diǎn)，主要存放元數(shù)據(jù)（meta）；賬本。(3)SecondaryNameNode稱為從元數(shù)據(jù)節(jié)點(diǎn)，是命名空間的冷備份（Namenode秘書），高可用（HA）解決方案，HDFS2.0采用另外一種機(jī)制。(4)DataNode稱為數(shù)據(jù)節(jié)點(diǎn)，是存放數(shù)據(jù)的物理空間，以塊（block）為基本單位；(5)NameNode和DataNode是主從結(jié)構(gòu)；(6)塊是HDFS操作最小單位，一般為128M；(7)元數(shù)據(jù)(meta)=文件目錄結(jié)構(gòu)信息(catalogue)+操作日志信息(log)，數(shù)據(jù)的描述信息(8)NameNode只存放catalogue，和SecondaryNameNode之間只相差3秒的catalogue；(9)DataNode按機(jī)架（rack）進(jìn)行組織，圖3.5有兩個(gè)機(jī)架；(10)客戶端只能同NameNode交互；2.2.1HDFS體系結(jié)構(gòu)2.2HDFS第二章大數(shù)據(jù)生態(tài)系統(tǒng)(11)一個(gè)數(shù)據(jù)塊通常要備份3份。第一份放到任務(wù)發(fā)起用戶所在節(jié)點(diǎn)，如果請求是集群外發(fā)起，就隨機(jī)選一不忙的節(jié)點(diǎn)；第二份放到與第一份不同的機(jī)架的節(jié)點(diǎn)上，第三份放到與第一份所在機(jī)架不同節(jié)點(diǎn)上，如圖中標(biāo)記黑色的塊；(12)DataNode定時(shí)向NameNode發(fā)送狀態(tài)信息（心跳，Heartbeats）,維護(hù)Block到本地文件系統(tǒng)（HDFSClient）的映射關(guān)系;(13)NameNode主要任務(wù)是：監(jiān)控心跳、負(fù)載平衡、數(shù)據(jù)塊備份的位置信息；(14)冷備份每3秒進(jìn)行一次，SecondaryNameNode冷備份過程見圖3.6；完成edits和fsimage同步，觸發(fā)冷備份的條件：3600秒或edits大于64M(15)NameNode結(jié)構(gòu)見圖3.7；(16)HDFS白話描述：如果把HDFS比作一本書，Namenode存儲(chǔ)的是書的目錄，Datanode存儲(chǔ)的就是書的正文內(nèi)容，一章是一個(gè)文件，一節(jié)是一個(gè)塊，目錄稱為元數(shù)據(jù)，目錄指明的各章節(jié)頁碼稱為映射，用戶訪問數(shù)據(jù)，首先要訪問Namenode；(17)HDFS優(yōu)點(diǎn)：采用抽象的塊概念可以帶來以下幾個(gè)明顯的好處：支持大規(guī)模文件存儲(chǔ)；簡化系統(tǒng)設(shè)計(jì)；適合數(shù)據(jù)備份；(18)HDFS缺點(diǎn)：塊的大小遠(yuǎn)遠(yuǎn)大于普通文件系統(tǒng)，所以不適合存儲(chǔ)小文件。2.2HDFS第二章大數(shù)據(jù)生態(tài)系統(tǒng)NameNode（主節(jié)點(diǎn)，其結(jié)構(gòu)見圖）2.2HDFS第二章大數(shù)據(jù)生態(tài)系統(tǒng)元數(shù)據(jù)細(xì)節(jié)2.2HDFS第二章大數(shù)據(jù)生態(tài)系統(tǒng)SecondaryNameNode（冷備份）冷備份過程如圖所示。在冷備份合并過程中發(fā)生的操作記錄在edits.new中。2.2HDFS第二章大數(shù)據(jù)生態(tài)系統(tǒng)2.2.2HDFS存儲(chǔ)原理(1)讀數(shù)據(jù)過程

類似圖書館借書。第1步：訪問圖書館網(wǎng)站（Namenode），提出請求；第2步：返回請求結(jié)果(Metadata)；第3不：去圖書館借書（Datanode）；第4步：拿到想借的書（Read）2.2HDFS第二章大數(shù)據(jù)生態(tài)系統(tǒng)2.2.2HDFS存儲(chǔ)原理(2)讀數(shù)據(jù)算法描述2.2HDFS第二章大數(shù)據(jù)生態(tài)系統(tǒng)2.2.2HDFS存儲(chǔ)原理(3)讀數(shù)據(jù)代碼實(shí)現(xiàn)importjava.io.BufferedReader;importjava.io.InputStreamReader;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.FileSystem;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.fs.FSDataInputStream;publicclassChapter3{publicstaticvoidmain(String[]args){try{Configurationconf=newConfiguration();conf.set("fs.defaultFS","hdfs://localhost:9000");conf.set("fs.hdfs.impl","org.apache.hadoop.hdfs.DistributedFileSystem");FileSystemfs=FileSystem.get(conf);Pathfile=newPath("test");FSDataInputStreamgetIt=fs.open(file);BufferedReaderd=newBufferedReader(newInputStreamReader(getIt));Stringcontent=d.readLine();//讀取文件一行System.out.println(content);d.close();//關(guān)閉文件fs.close();//關(guān)閉hdfs}catch(Exceptione){e.printStackTrace();}

}}}2.2HDFS第二章大數(shù)據(jù)生態(tài)系統(tǒng)2.2.2HDFS存儲(chǔ)原理(4)寫數(shù)據(jù)策略

寫數(shù)據(jù)包括增、刪、改，與讀數(shù)據(jù)過程不同。

第一個(gè)副本放置在上傳文件的數(shù)據(jù)節(jié)點(diǎn)；如果是集群外提交，則隨機(jī)挑選一臺(tái)磁盤不太滿、CPU不太忙的節(jié)點(diǎn)。

第二個(gè)副本放置在與第一個(gè)副本不同的機(jī)架的節(jié)點(diǎn)上。

第三個(gè)副本與第一個(gè)副本相同機(jī)架的其他節(jié)點(diǎn)上。

更多副本隨機(jī)節(jié)點(diǎn)2.2HDFS第二章大數(shù)據(jù)生態(tài)系統(tǒng)2.2.2HDFS存儲(chǔ)原理(5)寫數(shù)據(jù)算法描述2.2HDFS第二章大數(shù)據(jù)生態(tài)系統(tǒng)2.2.2HDFS存儲(chǔ)原理(6)寫數(shù)據(jù)代碼實(shí)現(xiàn)importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.FileSystem;importorg.apache.hadoop.fs.FSDataOutputStream;importorg.apache.hadoop.fs.Path;publicclassChapter3{publicstaticvoidmain(String[]args){try{Configurationconf=newConfiguration();conf.set("fs.defaultFS","hdfs://localhost:9000");conf.set("fs.hdfs.impl","org.apache.hadoop.hdfs.DistributedFileSystem");FileSystemfs=FileSystem.get(conf);byte[]buff="Helloworld".getBytes();//要寫入的內(nèi)容Stringfilename="test";//要寫入的文件名FSDataOutputStreamos=fs.create(newPath(filename));os.write(buff,0,buff.length);System.out.println("Create:"+filename);os.close();fs.close();}catch(Exceptione){e.printStackTrace();}2.2HDFS第二章大數(shù)據(jù)生態(tài)系統(tǒng)2.2.2HDFS存儲(chǔ)原理(7)數(shù)據(jù)出錯(cuò)與恢復(fù)HDFS具有較高的容錯(cuò)性，可以兼容廉價(jià)的硬件，它把硬件出錯(cuò)看作一種常態(tài)，而不是異常，并設(shè)計(jì)了相應(yīng)的機(jī)制檢測數(shù)據(jù)錯(cuò)誤和進(jìn)行自動(dòng)恢復(fù)，主要包括以下幾種情形：名稱節(jié)點(diǎn)出錯(cuò)、數(shù)據(jù)節(jié)點(diǎn)出錯(cuò)和數(shù)據(jù)出錯(cuò)。①名稱節(jié)點(diǎn)出錯(cuò)

名稱節(jié)點(diǎn)保存了所有的元數(shù)據(jù)信息，其中，最核心的兩大數(shù)據(jù)結(jié)構(gòu)是FsImage和Editlog，如果這兩個(gè)文件發(fā)生損壞，那么整個(gè)HDFS實(shí)例將失效。因此，HDFS設(shè)置了備份機(jī)制，把這些核心文件同步復(fù)制到備份服務(wù)器SecondaryNameNode上。當(dāng)名稱節(jié)點(diǎn)出錯(cuò)時(shí)，就可以根據(jù)備份服務(wù)器SecondaryNameNode中的FsImage和Editlog數(shù)據(jù)進(jìn)行恢復(fù)。②數(shù)據(jù)節(jié)點(diǎn)出錯(cuò)

每個(gè)數(shù)據(jù)節(jié)點(diǎn)會(huì)定期向名稱節(jié)點(diǎn)發(fā)送“心跳”信息，向名稱節(jié)點(diǎn)報(bào)告自己的狀態(tài)

當(dāng)數(shù)據(jù)節(jié)點(diǎn)發(fā)生故障，或者網(wǎng)絡(luò)發(fā)生斷網(wǎng)時(shí)，名稱節(jié)點(diǎn)就無法收到來自一些數(shù)據(jù)節(jié)點(diǎn)的心跳信息，這時(shí)，這些數(shù)據(jù)節(jié)點(diǎn)就會(huì)被標(biāo)記為“宕機(jī)”，節(jié)點(diǎn)上面的所有數(shù)據(jù)都會(huì)被標(biāo)記為“不可讀”，名稱節(jié)點(diǎn)不會(huì)再給它們發(fā)送任何I/O請求。這時(shí)，有可能出現(xiàn)一種情形，即由于一些數(shù)據(jù)節(jié)點(diǎn)的不可用，會(huì)導(dǎo)致一些數(shù)據(jù)塊的副本數(shù)量小于冗余因子。

名稱節(jié)點(diǎn)會(huì)定期檢查這種情況，一旦發(fā)現(xiàn)某個(gè)數(shù)據(jù)塊的副本數(shù)量小于冗余因子，就會(huì)啟動(dòng)數(shù)據(jù)冗余復(fù)制，為它生成新的副本。HDFS和其它分布式文件系統(tǒng)的最大區(qū)別就是可以調(diào)整冗余數(shù)據(jù)的位置。

2.2HDFS第二章大數(shù)據(jù)生態(tài)系統(tǒng)2.2.2HDFS存儲(chǔ)原理(7)數(shù)據(jù)出錯(cuò)與恢復(fù)③數(shù)據(jù)出錯(cuò)

網(wǎng)絡(luò)傳輸和磁盤錯(cuò)誤等因素，都會(huì)造成數(shù)據(jù)錯(cuò)誤。

客戶端在讀取到數(shù)據(jù)后，會(huì)采用md5和sha1對數(shù)據(jù)塊進(jìn)行校驗(yàn)，以確定讀取到正確的數(shù)據(jù)。

在文件被創(chuàng)建時(shí)，客戶端就會(huì)對每一個(gè)文件塊進(jìn)行信息摘錄，并把這些信息寫入到同一個(gè)路徑的隱藏文件里面。

當(dāng)客戶端讀取文件的時(shí)候，會(huì)先讀取該信息文件，然后，利用該信息文件對每個(gè)讀取的數(shù)據(jù)塊進(jìn)行校驗(yàn)，如果校驗(yàn)出錯(cuò)，客戶端就會(huì)請求到另外一個(gè)數(shù)據(jù)節(jié)點(diǎn)讀取該文件塊，并且向名稱節(jié)點(diǎn)報(bào)告這個(gè)文件塊有錯(cuò)誤，名稱節(jié)點(diǎn)會(huì)定期檢查并且重新復(fù)制這個(gè)塊。第二章大數(shù)據(jù)生態(tài)系統(tǒng)2.3

MapReduce2.1認(rèn)識(shí)Hadoop2.4

zookeeper2.2

HDFSContents2.3.12.3.2實(shí)例分析：WordCount2.3.32.3.4MapReduce體系結(jié)構(gòu)MapReduce工作流程MapReduce編程實(shí)踐分布式并行編程MapReduce模型簡介2.3.1MapReduce體系結(jié)構(gòu)Map和Reduce函數(shù)問題：在MapReduce出現(xiàn)之前，已經(jīng)有并行計(jì)算框架了，那么為什么Google還需要MapReduce？MapReduce與傳統(tǒng)的并行計(jì)算框架相比有什么優(yōu)勢？一、分布式并行編程回顧傳統(tǒng)并行計(jì)算框架MapReduce集群架構(gòu)/容錯(cuò)性共享式(共享內(nèi)存/共享存儲(chǔ))，容錯(cuò)性差非共享式，容錯(cuò)性好硬件/價(jià)格/擴(kuò)展性刀片服務(wù)器、高速網(wǎng)、SAN，價(jià)格貴，擴(kuò)展性差普通PC機(jī)，便宜，擴(kuò)展性好編程/學(xué)習(xí)難度what-how，難what，簡單適用場景實(shí)時(shí)、細(xì)粒度計(jì)算、計(jì)算密集型批處理、非實(shí)時(shí)、數(shù)據(jù)密集型分布式并行編程MapReduce模型簡介2.3.2MapReduce體系結(jié)構(gòu)Map和Reduce函數(shù)輸入和輸出都是鍵值對二、MapReduce模型簡介2、一個(gè)例子二、MapReduce模型簡介（1）MapReduce將復(fù)雜的、運(yùn)行于大規(guī)模集群上的并行計(jì)算過程高度地抽象為兩個(gè)函數(shù)：Map和Reduce（2）編程容易，不需要掌握分布式并行編程細(xì)節(jié)，也可以很容易把自己的程序運(yùn)行在分布式系統(tǒng)上，完成海量數(shù)據(jù)的計(jì)算（3）MapReduce采用“分而治之”策略，一個(gè)存儲(chǔ)在分布式文件系統(tǒng)中的大規(guī)模數(shù)據(jù)集，會(huì)被切分成許多獨(dú)立的分片（split），這些分片可以被多個(gè)Map任務(wù)并行處理（4）MapReduce設(shè)計(jì)的一個(gè)理念就是“計(jì)算向數(shù)據(jù)靠攏”，而不是“數(shù)據(jù)向計(jì)算靠攏”，因?yàn)?，移?dòng)數(shù)據(jù)需要大量的網(wǎng)絡(luò)傳輸開銷（5）MapReduce框架采用了Master/Slave架構(gòu)，包括一個(gè)Master和若干個(gè)Slave。Master上運(yùn)行JobTracker，Slave上運(yùn)行TaskTracker

2、基本思想3、MapReduce的體系結(jié)構(gòu)MapReduce體系結(jié)構(gòu)主要由四個(gè)部分組成，分別是：Client、JobTracker、TaskTracker以及Task二、MapReduce模型簡介1）Client用戶編寫的MapReduce程序通過Client提交到JobTracker端用戶可通過Client提供的一些接口查看作業(yè)運(yùn)行狀態(tài)2）JobTrackerJobTracker負(fù)責(zé)資源監(jiān)控和作業(yè)調(diào)度JobTracker監(jiān)控所有TaskTracker與Job的健康狀況，一旦發(fā)現(xiàn)失敗，就將相應(yīng)的任務(wù)轉(zhuǎn)移到其他節(jié)點(diǎn)JobTracker會(huì)跟蹤任務(wù)的執(zhí)行進(jìn)度、資源使用量等信息，并將這些信息告訴任務(wù)調(diào)度器（TaskScheduler），而調(diào)度器會(huì)在資源出現(xiàn)空閑時(shí)，選擇合適的任務(wù)去使用這些資源二、MapReduce模型簡介3）TaskTrackerTaskTracker會(huì)周期性地通過“心跳”將本節(jié)點(diǎn)上資源的使用情況和任務(wù)的運(yùn)行進(jìn)度匯報(bào)給JobTracker，同時(shí)接收J(rèn)obTracker發(fā)送過來的命令并執(zhí)行相應(yīng)的操作（如啟動(dòng)新任務(wù)、殺死任務(wù)等）TaskTracker使用“slot”等量劃分本節(jié)點(diǎn)上的資源量（CPU、內(nèi)存等）。一個(gè)Task獲取到一個(gè)slot后才有機(jī)會(huì)運(yùn)行，而Hadoop調(diào)度器的作用就是將各個(gè)TaskTracker上的空閑slot分配給Task使用。slot分為Mapslot和Reduceslot兩種，分別供MapTask和ReduceTask使用4）TaskTask分為MapTask和ReduceTask兩種，均由TaskTracker啟動(dòng)二、MapReduce模型簡介分布式并行編程MapReduce模型簡介2.3.2MapReduce體系結(jié)構(gòu)Map和Reduce函數(shù)程序WordCount輸入一個(gè)包含大量單詞的文本文件輸出文件中每個(gè)單詞及其出現(xiàn)次數(shù)（頻數(shù)），并按照單詞字母順序排序，每個(gè)單詞和其頻數(shù)占一行，單詞和頻數(shù)之間有間隔輸入輸出HelloWorldHelloHadoopHelloMapReduceHadoop1Hello3MapReduce1World1三、Map和Reduce函數(shù)

一個(gè)WordCount執(zhí)行過程的實(shí)例三、Map和Reduce函數(shù)三、Map和Reduce函數(shù)三、Map和Reduce函數(shù)函數(shù)輸入輸出說明Map<k1,v1>如：<行號(hào),”abc”>List(<k2,v2>)如：<“a”,1><“b”,1><“c”,1>1.將小數(shù)據(jù)集進(jìn)一步解析成一批<key,value>對，輸入Map函數(shù)中進(jìn)行處理2.每一個(gè)輸入的<k1,v1>會(huì)輸出一批<k2,v2>。<k2,v2>是計(jì)算的中間結(jié)果Reduce<k2,List(v2)>如：<“a”,<1,1,1>><k3,v3><“a”,3>輸入的中間結(jié)果<k2,List(v2)>中的List(v2)表示是一批屬于同一個(gè)k2的value2編寫Map處理邏輯Map輸入為<偏移量，一行文本>,范型<LongWritable,Text>Map輸出為<單詞，1>,范型<Text,IntWritable>2編寫Map處理邏輯publicclassWordCountMapperextendsMapper<LongWritable,Text,Text,IntWritable>{

//對數(shù)據(jù)進(jìn)行打散

@Override

protectedvoidmap(LongWritablekey,Textvalue,Contextcontext)throwsIOException,InterruptedException{

//1.接入數(shù)據(jù)hellorebahellomimi

Stringline=value.toString();

//2.對數(shù)據(jù)進(jìn)行切分

String[]words=line.split("");

//3.寫出以<hello,1>

for(Stringw:words){

//寫出reducer端

context.write(newText(w),newIntWritable(1));

}

}

}3編寫Reduce處理邏輯Reduce的輸入數(shù)據(jù)為<”China”,<1,1,1>>

，范型<Text,Iterable容器>

Reduce輸出數(shù)據(jù)為<”China”,3>

，范型<Text,IntWritable>publicclassWordCountReducerextendsReducer<Text,IntWritable,Text,IntWritable>{//key->單詞values->次數(shù)11111@Overrideprotectedvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{//1.記錄出現(xiàn)的次數(shù)

intsum=0;for(IntWritablev:values){sum+=v.get();}//2.累加求和輸出

context.write(key,newIntWritable(sum));}}7.6.3編寫Reduce處理邏輯工作流程概述MapReduce各個(gè)執(zhí)行階段2.3.3MapReduce工作流程Shuffle過程詳解應(yīng)用程序執(zhí)行過程3.1.2阿里云一、工作流程概述不同的Map任務(wù)之間不會(huì)進(jìn)行通信不同的Reduce任務(wù)之間也不會(huì)發(fā)生任何信息交換用戶不能顯式地從一臺(tái)機(jī)器向另一臺(tái)機(jī)器發(fā)送消息所有的數(shù)據(jù)交換都是通過MapReduce框架自身去實(shí)現(xiàn)的工作流程概述MapReduce各個(gè)執(zhí)行階段2.3.3MapReduce工作流程Shuffle過程詳解應(yīng)用程序執(zhí)行過程3.1.2阿里云二、MapReduce各個(gè)執(zhí)行階段HDFS以固定大小的block為基本單位存儲(chǔ)數(shù)據(jù)，而對于MapReduce而言，其處理單位是split。split是一個(gè)邏輯概念，它只包含一些元數(shù)據(jù)信息，比如數(shù)據(jù)起始位置、數(shù)據(jù)長度、數(shù)據(jù)所在節(jié)點(diǎn)等。它的劃分方法完全由用戶自己決定。關(guān)于Split（分片）3.1.2阿里云二、MapReduce各個(gè)執(zhí)行階段Reduce任務(wù)的數(shù)量最優(yōu)的Reduce任務(wù)個(gè)數(shù)取決于集群中可用的reduce任務(wù)槽(slot)的數(shù)目通常設(shè)置比reduce任務(wù)槽數(shù)目稍微小一些的Reduce任務(wù)個(gè)數(shù)（這樣可以預(yù)留一些系統(tǒng)資源處理可能發(fā)生的錯(cuò)誤）Map任務(wù)的數(shù)量Hadoop為每個(gè)split創(chuàng)建一個(gè)Map任務(wù)，split的多少?zèng)Q定了Map任務(wù)的數(shù)目。大多數(shù)情況下，理想的分片大小是一個(gè)HDFS塊3.1.2阿里云二、MapReduce各個(gè)執(zhí)行階段工作流程概述MapReduce各個(gè)執(zhí)行階段2.3.2MapReduce工作流程Shuffle過程詳解應(yīng)用程序執(zhí)行過程1.Shuffle過程簡介3.1.2阿里云三、Shuffle過程詳解2.Map端的Shuffle過程每個(gè)Map任務(wù)分配一個(gè)緩存MapReduce默認(rèn)100MB緩存設(shè)置溢寫比例0.8分區(qū)默認(rèn)采用哈希函數(shù)排序是默認(rèn)的操作排序后可以合并（Combine）合并不能改變最終結(jié)果在Map任務(wù)全部結(jié)束之前進(jìn)行歸并歸并得到一個(gè)大的文件，放在本地磁盤文件歸并時(shí)，如果溢寫文件數(shù)量大于預(yù)定值（默認(rèn)是3）則可以再次啟動(dòng)Combiner，少于3不需要JobTracker會(huì)一直監(jiān)測Map任務(wù)的執(zhí)行，并通知Reduce任務(wù)來領(lǐng)取數(shù)據(jù)合并（Combine）和歸并（Merge）的區(qū)別：兩個(gè)鍵值對<“a”,1>和<“a”,1>，如果合并，會(huì)得到<“a”,2>，如果歸并，會(huì)得到<“a”,<1,1>>3.1.2阿里云三、Shuffle過程詳解3.Reduce端的Shuffle過程Reduce端的Shuffle過程Reduce任務(wù)通過RPC向JobTracker詢問Map任務(wù)是否已經(jīng)完成，若完成，則領(lǐng)取數(shù)據(jù)Reduce領(lǐng)取數(shù)據(jù)先放入緩存，來自不同Map機(jī)器，先歸并，再合并，寫入磁盤多個(gè)溢寫文件歸并成一個(gè)或多個(gè)大文件，文件中的鍵值對是排序的當(dāng)數(shù)據(jù)很少時(shí)，不需要溢寫到磁盤，直接在緩存中歸并，然后輸出給Reduce3.1.2阿里云三、Shuffle過程詳解工作流程概述MapReduce各個(gè)執(zhí)行階段2.3.2MapReduce工作流程Shuffle過程詳解應(yīng)用程序執(zhí)行過程3.1.2阿里云四、MapReduce應(yīng)用程序執(zhí)行過程應(yīng)用場景自然連接2.3.4MapReduce編程實(shí)踐實(shí)戰(zhàn)一、MapReduce應(yīng)用場景MapReduce可以很好地應(yīng)用于各種計(jì)算問題關(guān)系代數(shù)運(yùn)算（選擇、投影、并、交、差、連接）分組與聚合運(yùn)算矩陣-向量乘法矩陣乘法應(yīng)用場景自然連接2.3.4MapReduce編程實(shí)踐實(shí)戰(zhàn)假設(shè)有關(guān)系R(A，B)和S(B,C)，對二者進(jìn)行自然連接操作使用Map過程，把來自R的每個(gè)元組<a,b>轉(zhuǎn)換成一個(gè)鍵值對<b,<R,a>>，其中的鍵就是屬性B的值。把關(guān)系R包含到值中，這樣做使得我們可以在Reduce階段，只把那些來自R的元組和來自S的元組進(jìn)行匹配。類似地，使用Map過程，把來自S的每個(gè)元組<b,c>，轉(zhuǎn)換成一個(gè)鍵值對<b,<S,c>>所有具有相同B值的元組被發(fā)送到同一個(gè)Reduce進(jìn)程中，Reduce進(jìn)程的任務(wù)是，把來自關(guān)系R和S的、具有相同屬性B值的元組進(jìn)行合并Reduce進(jìn)程的輸出則是連接后的元組<a,b,c>，輸出被寫到一個(gè)單獨(dú)的輸出文件中二、用MapReduce實(shí)現(xiàn)關(guān)系的自然連接用MapReduce實(shí)現(xiàn)關(guān)系的自然連接應(yīng)用場景自然連接2.3.4MapReduce編程實(shí)踐實(shí)戰(zhàn)三、MapReduce實(shí)踐1 任務(wù)要求2 編寫Map處理邏輯3 編寫Reduce處理邏輯4 編寫main方法5 編譯打包代碼以及運(yùn)行程序1任務(wù)要求文件A的內(nèi)容如下：ChinaismymotherlandIloveChina文件B的內(nèi)容如下：IamfromChina期望結(jié)果如右側(cè)所示：I2is1China3my1love1am1from1motherland12編寫Map處理邏輯Map輸入類型為<key,value>期望的Map輸出類型為<單詞，1>2編寫Map處理邏輯publicclassWordCountMapperextendsMapper<LongWritable,Text,Text,IntWritable>{

//對數(shù)據(jù)進(jìn)行打散

@Override

protectedvoidmap(LongWritablekey,Textvalue,Contextcontext)throwsIOException,InterruptedException{

//1.接入數(shù)據(jù)hellorebahellomimi

Stringline=value.toString();

//2.對數(shù)據(jù)進(jìn)行切分

String[]words=line.split("");

//3.寫出以<hello,1>

for(Stringw:words){

//寫出reducer端

context.write(newText(w),newIntWritable(1));

}

}

}3編寫Reduce處理邏輯在Reduce處理數(shù)據(jù)之前，Map的結(jié)果首先通過Shuffle階段進(jìn)行整理Reduce階段的任務(wù)：對輸入數(shù)字序列進(jìn)行求和Reduce的輸入數(shù)據(jù)為<key,Iterable容器>Reduce任務(wù)的輸入數(shù)據(jù)：<”I”,<1,1>><”is”,1>……<”from”,1><”China”,<1,1,1>>

publicclassWordCountReducerextendsReducer<Text,IntWritable,Text,IntWritable>{//key->單詞values->次數(shù)11111@Overrideprotectedvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{//1.記錄出現(xiàn)的次數(shù)

intsum=0;for(IntWritablev:values){sum+=v.get();}//2.累加求和輸出

context.write(key,newIntWritable(sum));}}7.6.3編寫Reduce處理邏輯4編寫main方法publicclassWordCountDriver{publicstaticvoidmain(String[]args)throwsIOException,ClassNotFoundException,InterruptedException{Configurationconf=newConfiguration();Jobjob=Job.getInstance(conf);//1.創(chuàng)建job任務(wù)

job.setJarByClass(WordCountDriver.class);//2.指定jar包位置

job.setMapperClass(WordCountMapper.class);//3.關(guān)聯(lián)使用的Mapper類

job.setReducerClass(WordCountReducer.class);//4.關(guān)聯(lián)使用的Reducer類

job.setMapOutputKeyClass(Text.class);//5.設(shè)置mapper階段輸出的數(shù)據(jù)類型

job.setMapOutputValueClass(IntWritable.class);

job.setOutputKeyClass(Text.class);//6.設(shè)置reducer階段輸出的數(shù)據(jù)類型

job.setOutputValueClass(IntWritable.class);

FileInputFormat.setInputPaths(job,newPath(args[0]));//7.設(shè)置數(shù)據(jù)輸入的路徑

FileOutputFormat.setOutputPath(job,newPath(args[1]));//8.設(shè)置數(shù)據(jù)輸出的路徑

boolean

rs=job.waitForCompletion(true);//9.提交任務(wù)

System.exit(rs?0:1);}}5編譯打包代碼以及運(yùn)行程序?qū)嶒?yàn)步驟：使用java編譯程序，生成.class文件將.class文件打包為jar包運(yùn)行jar包（需要啟動(dòng)Hadoop）查看結(jié)果Hadoop2.x版本中的依賴jarHadoop2.x版本中jar不再集中在一個(gè)hadoop-core*.jar中，而是分成多個(gè)jar，如使用Hadoop2.6.0運(yùn)行WordCount實(shí)例至少需要如下三個(gè)jar:$HADOOP_HOME/share/hadoop/common/hadoop-common-2.6.0.jar$HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-client-core-2.6.0.jar$HADOOP_HOME/share/hadoop/common/lib/commons-cli-1.2.jar通過命令

hadoopclasspath

可以得到運(yùn)行Hadoop程序所需的全部classpath信息將Hadoop的classhpath信息添加到CLASSPATH變量中，在~/.bashrc中增加如下幾行：exportHADOOP_HOME=/usr/local/hadoopexportCLASSPATH=$($HADOOP_HOME/bin/hadoopclasspath):$CLASSPATH

執(zhí)行

source~/.bashrc

使變量生效，接著就可以通過

javac

命令編譯WordCount.java接著把.class文件打包成jar，才能在Hadoop中運(yùn)行：運(yùn)行程序：第二章大數(shù)據(jù)生態(tài)系統(tǒng)2.4zookeeper2.1認(rèn)識(shí)Hadoop2.3

MapReduce2.2

HDFS什么是zookeeperzookeeper功能2.4.1Zookeeper概述zookeeper結(jié)構(gòu)第二章大數(shù)據(jù)生態(tài)系統(tǒng)2.4.1zookeeper概述一、問題提出HDFSMapReduceHBASE如果HDFS、MapReduce……的主節(jié)點(diǎn)出故障怎么辦？2.4.1zookeeper概述第二章大數(shù)據(jù)生態(tài)系統(tǒng)分布式集群協(xié)調(diào)器二、什么是zookeeper配置管理命名服務(wù)提供分布式同步集群管理第二章大數(shù)據(jù)生態(tài)系統(tǒng)2.4.1zookeeper概述三、zookeeper能做什么第二章大數(shù)據(jù)生態(tài)系統(tǒng)2.4.1zookeeper概述1、命名服務(wù)：一個(gè)分布式系統(tǒng)的各節(jié)點(diǎn)可能需要一個(gè)規(guī)范的、各節(jié)點(diǎn)的唯一的命名（例如節(jié)點(diǎn)名、CPU編號(hào)等），ZK可以實(shí)現(xiàn)這個(gè)的應(yīng)用場景。2、配置管理：各個(gè)節(jié)點(diǎn)也會(huì)有一致的信息，例如每個(gè)節(jié)點(diǎn)的主配置信息。3、提供分布式同步：

為了管理方便，新加入的節(jié)點(diǎn)也需要快速同步這些信息。4、集群管理:

(1)集群狀態(tài)監(jiān)控和通知：分布式系統(tǒng)中的每個(gè)節(jié)點(diǎn)需要知道整個(gè)系統(tǒng)的狀態(tài)、知道系統(tǒng)中每個(gè)節(jié)點(diǎn)的狀態(tài)：當(dāng)有新節(jié)點(diǎn)加入時(shí)它需要知道、當(dāng)有節(jié)點(diǎn)出現(xiàn)故障時(shí)它需要知道、當(dāng)有節(jié)點(diǎn)退出時(shí)它需要知道。ZK就是這樣一個(gè)“通知工具”。(2)協(xié)調(diào)資源搶占（鎖）：當(dāng)分布式系統(tǒng)的多個(gè)節(jié)點(diǎn)試圖同時(shí)搶占唯一資源時(shí)（例如同時(shí)寫入一個(gè)文件），就需要對這個(gè)唯一資源的使用進(jìn)行協(xié)調(diào)。這是ZK的“協(xié)調(diào)者”功能。(3)分派計(jì)算任務(wù)：如何協(xié)調(diào)1000個(gè)需要同時(shí)處理的任務(wù)到分布式系統(tǒng)的13個(gè)節(jié)點(diǎn)？如何保證執(zhí)行失敗的任務(wù)能被重新執(zhí)行？如何在某個(gè)節(jié)點(diǎn)崩潰的情況下，接管其正在處理的任務(wù)？配置管理——訂閱/發(fā)布訂閱者（客戶端）通過一定的方法告訴Zookeeper服務(wù)器，自己對哪個(gè)節(jié)點(diǎn)的數(shù)據(jù)感興趣，那么在服務(wù)端數(shù)據(jù)發(fā)生變化時(shí)，就會(huì)