網(wǎng)盤(pán)測(cè)試數(shù)據(jù)mr原理與編程

1、第3章MR原理與編程講師： 凌宏博郵箱： M/R計(jì)算模型什么M/R模型MapReduce 是Hadoop 的核心組成, 用于大規(guī)模數(shù)據(jù)集的并行計(jì)算。概念”Map（映射）” 和 “Reduce（歸并）” 即表示將海量數(shù)據(jù)映射成多份文件，在各自節(jié)點(diǎn)并行處理，再將處理結(jié)果發(fā)送給Reduce節(jié)點(diǎn) 進(jìn)行歸并計(jì)算，可以簡(jiǎn)單的表示為：map: (K1,V1) list（K2,V2）reduce: (K2,list(V2) list(K3,V3)Hadoop M/R架構(gòu)同HDFS一樣，其也采用了M/S 架構(gòu)，具體如圖所示。它主要由以下幾個(gè)組件組成：Client、JobTracker、TaskTracker和T

2、askM/R計(jì)算模型M/R組成架構(gòu)M/R計(jì)算模型M/R組件簡(jiǎn)介1. client用戶(hù)編寫(xiě)的M/R程序通過(guò)clientd提交到JobTracker, Hadoop內(nèi)部用作業(yè)（job）代表M/R程序2. JobTracker主要負(fù)責(zé)自由監(jiān)控和作業(yè)調(diào)度。監(jiān)控所有TaskTracker與作業(yè)的狀況，一旦發(fā)現(xiàn)失敗，其會(huì)將相應(yīng)的任務(wù)轉(zhuǎn)移到其他節(jié)點(diǎn)；同時(shí)會(huì)跟任務(wù)的進(jìn)度、資源使用等信息，并將此信息告訴任務(wù)調(diào)度器3. TaskTracker會(huì)周期性的通過(guò)Heartbeat將本節(jié)點(diǎn)上資源的使用情況和任務(wù)運(yùn)行進(jìn)度匯報(bào)給JobTracker,同時(shí)接受JobTracker發(fā)過(guò)來(lái)的命令（如啟動(dòng)新任務(wù)、殺死任務(wù)等）。Tas

3、kTracker使用“slot”來(lái)劃分本節(jié)點(diǎn)上的資源量。“slot”代表計(jì)算資源（CPU、內(nèi)存等）。一個(gè)Task獲取到一個(gè)slot后才有機(jī)會(huì)運(yùn)行，而Hadoop調(diào)度器的作用就是將各個(gè)TaskTracker上空閑的slot分配給Task使用。Slot分為Map slot 和 Reduce slot兩種， TaskTracker可以通過(guò)限制slot數(shù)目（可配參數(shù)）來(lái)限定Task的并發(fā)度4. Task分為Map Task 和 Reduce Task，均有TaskTracker啟動(dòng)。從上一節(jié)我們知道，HDFS以固定大小的block為基本存儲(chǔ)單位，而對(duì)于MapReduce而言，其處理單位是split，s

4、plit是一個(gè)邏輯上的概念，它包含一些元信息，比如數(shù)據(jù)起始位置、長(zhǎng)度、所在節(jié)點(diǎn)等。它的劃分方法由用戶(hù)自己決定，但需要注意的是有多少個(gè)split就有多少個(gè)MapTask數(shù)目，因?yàn)槊總€(gè)split會(huì)交由一個(gè)MapTask處理M/R計(jì)算模型M/R任務(wù)執(zhí)行流程分析M/R計(jì)算模型Map Task執(zhí)行流程分析1. TaskNode節(jié)點(diǎn)啟動(dòng)分配JobTracker個(gè)Map Task任務(wù)，啟用InputFormat對(duì)象(具體為RecordReader)讀取分配給本節(jié)點(diǎn)split分片，將每一行的數(shù)據(jù)讀取成key/value鍵值對(duì)2. mapper程序的map()函數(shù)會(huì)接收key/value鍵值對(duì)的數(shù)據(jù)，做邏輯處理

5、，然后將之輸出3.開(kāi)始MapReduce的shuffle過(guò)程，該過(guò)程的性能的好壞可以直接決定MapReduce的性能。)mapper程序會(huì)將經(jīng)處理過(guò)的key/value數(shù)據(jù)先加入該Task的環(huán)形的內(nèi)存緩沖區(qū)，直到達(dá)到緩存達(dá)到閥值，然后會(huì)將數(shù)據(jù)spill到linux的本地磁盤(pán)上(注意，這些中間數(shù)據(jù)不會(huì)被保存到HDFS)，生成一個(gè)磁盤(pán)文件4. mapper處理玩自己分片的數(shù)據(jù)后，會(huì)將磁盤(pán)上的所有spill文件合并5.當(dāng)一個(gè)Map Task任務(wù)處理完后，TaskTracker節(jié)點(diǎn)會(huì)告訴JobTracker節(jié)點(diǎn)任務(wù)執(zhí)行完畢并等待新的任務(wù)，然后JobTracker會(huì)告知reducer程序，區(qū)獲取屬于re

6、ducer自己的數(shù)據(jù)M/R計(jì)算模型Reduce Task執(zhí)行流程分析1. reducer通過(guò)多線(xiàn)程(默認(rèn)為5個(gè)copy線(xiàn)程)開(kāi)始到已執(zhí)行完成的mapper節(jié)點(diǎn)上copy屬于自己的partition數(shù)據(jù)2. reducer會(huì)先將copy的數(shù)據(jù)先放在本機(jī)的內(nèi)存緩存中，合并拷貝數(shù)據(jù)，當(dāng)緩存中放不下copy數(shù)據(jù)時(shí)，會(huì)將內(nèi)叢中的的數(shù)據(jù)刷新到磁盤(pán)上，數(shù)據(jù)copy完成后，會(huì)將雖有數(shù)據(jù)合并成一個(gè)大文件，數(shù)據(jù)格式由原來(lái)的變成3.將合并后的文件排序4.接下來(lái)reducer開(kāi)始執(zhí)行業(yè)務(wù)邏輯處理的reduce()方法，并將數(shù)據(jù)最終輸出到HDFS上5.所有的Task都執(zhí)行完成后，整個(gè)處理過(guò)程完成M/R計(jì)算模型Reduc

7、e Task執(zhí)行流程分析1. reducer通過(guò)多線(xiàn)程(默認(rèn)為5個(gè)copy線(xiàn)程)開(kāi)始到已執(zhí)行完成的mapper節(jié)點(diǎn)上copy屬于自己的partition數(shù)據(jù)2. reducer會(huì)先將copy的數(shù)據(jù)先放在本機(jī)的內(nèi)存緩存中，合并拷貝數(shù)據(jù)，當(dāng)緩存中放不下copy數(shù)據(jù)時(shí)，會(huì)將內(nèi)叢中的的數(shù)據(jù)刷新到磁盤(pán)上，數(shù)據(jù)copy完成后，會(huì)將雖有數(shù)據(jù)合并成一個(gè)大文件，數(shù)據(jù)格式由原來(lái)的變成3.將合并后的文件排序4.接下來(lái)reducer開(kāi)始執(zhí)行業(yè)務(wù)邏輯處理的reduce()方法，并將數(shù)據(jù)最終輸出到HDFS上5.所有的Task都執(zhí)行完成后，整個(gè)處理過(guò)程完成M/R編程動(dòng)手?jǐn)U展一個(gè)簡(jiǎn)單程序練習(xí)：統(tǒng)計(jì)一組文檔中每個(gè)單詞出現(xiàn)的次數(shù)

8、思考：1. 如果該文檔很小，一個(gè)簡(jiǎn)單的程序即可完成工作2. 如果文檔數(shù)量激增，如何處理？ 例如：編寫(xiě)一個(gè)垃圾郵件過(guò)濾器，獲取上百萬(wàn)封垃圾郵件中經(jīng)常使用的單詞 方案一：?jiǎn)闻_(tái)機(jī)器反復(fù)遍歷所有文檔，非常耗時(shí)，提出方案二。 方案二：第一階段多臺(tái)機(jī)器處理不同文檔內(nèi)容，當(dāng)所有機(jī)器都處理完事， 啟動(dòng)第二階段合并這些結(jié)果不是很難對(duì)嗎？ 考慮細(xì)節(jié): (文件的存放、機(jī)器的內(nèi)存、第二階段機(jī)器的瓶頸)1.單詞統(tǒng)計(jì)程序正在變得復(fù)雜，為了滿(mǎn)足計(jì)算的要求，需要添加以下功能1. 文件存儲(chǔ)到多臺(tái)機(jī)器2. 內(nèi)存不夠時(shí)，將內(nèi)容寫(xiě)到磁盤(pán)3. 劃分第一個(gè)階段的中間數(shù)據(jù)4. 洗牌這些數(shù)據(jù)到第二階合適的段機(jī)器5. 容錯(cuò)機(jī)制 M/R編程M/

9、R編程入門(mén):wordcount一個(gè)簡(jiǎn)單的單詞統(tǒng)計(jì)在，當(dāng)數(shù)據(jù)量激增時(shí)，變得如此復(fù)雜，因此我們迫切需要一個(gè)計(jì)算框架，來(lái)幫我們管理和實(shí)現(xiàn)所有與可擴(kuò)展性相關(guān)的底層問(wèn)題。Hadoop的M/R編程模型就為我們提供了這么一個(gè)框架，它是我們不需要去了解底層分布式系統(tǒng)的細(xì)節(jié)，就能幫我們寫(xiě)好一個(gè)分布式應(yīng)用程序運(yùn)行：Hadoo自動(dòng)的example包下的：wordcount; 分析它的輸入和輸出，理解它的M/R執(zhí)行流程WordCount處理過(guò)程分析：1. 將文件拆分成splits，由于測(cè)試用的文件較小，所以每個(gè)文件為一個(gè)split，并將文件按行分割形成對(duì)，如圖所示。 M/R編程WordCount處理過(guò)程分析 2.將分

10、割好的對(duì)交給用戶(hù)定義的map方法進(jìn)行處理，生成新的對(duì)，如圖所示:M/R編程WordCount處理過(guò)程分析 3. 得到map方法輸出的對(duì)后，Mapper會(huì)將它們按照key值進(jìn)行排序，并執(zhí)行Combine過(guò)程，將key至相同value值累加，得到Mapper的最終輸出結(jié)果。如圖所示。M/R編程WordCount處理過(guò)程分析 4. Reducer先對(duì)從Mapper接收的數(shù)據(jù)進(jìn)行排序，再交由用戶(hù)自定義的reduce方法進(jìn)行處理，得到新的對(duì)，并作為WordCount的輸出結(jié)果，如圖所示M/R編程WordCount處理過(guò)程分析 Hadoop數(shù)據(jù)類(lèi)型序列化的數(shù)據(jù)類(lèi)型hadoop處理的文件通?？梢杂面I/值對(duì)來(lái)

11、表示，但Map/Reduce框架并不允許我們盡管在是任意類(lèi)。例如我們經(jīng)常把某些鍵與值稱(chēng)為整數(shù)、字符串等，但它們實(shí)際并不是Integer,String等標(biāo)準(zhǔn)java類(lèi)，因?yàn)闉榱俗屾I/值對(duì)可以在集群上移動(dòng)，M/R框架提供了一種序列化鍵/值對(duì)的方法。序列化的概念序列化（Serialization）是指把結(jié)構(gòu)化對(duì)象轉(zhuǎn)化為字節(jié)流。反序列化（Deserialization）是序列化的逆過(guò)程。即把字節(jié)流轉(zhuǎn)回結(jié)構(gòu)化對(duì)象Java序列化（java.io.Serializable）序列化在分布式的作用：進(jìn)程間通信，永久存儲(chǔ)。由于hadoop是分布式的系統(tǒng)，所以hadoop中的鍵和值，都應(yīng)該序列化 Hadoop數(shù)據(jù)類(lèi)

12、型Writable接口如果要實(shí)現(xiàn)序列化，只需實(shí)現(xiàn)Writable接口即可；其接口有兩個(gè)方法：void write(DataOutput out)void readFields(DataInput in)write方法是將對(duì)象屬性序列化到DataOutput中去，readFields方法則是將DataInput中的數(shù)據(jù)反序列化到對(duì)象屬性中 parable接口此接口在實(shí)現(xiàn)序列化的同時(shí)，還實(shí)現(xiàn)了比較功能。 它是Writable 和 parable接口的組合。對(duì)于hadoop中的鍵來(lái)說(shuō)，我們需要比較接口，因?yàn)樗麄儗⒃趓educe階段進(jìn)行排序比較 而值只需要被簡(jiǎn)單的傳遞即可Hadoop帶有一下預(yù)定義的類(lèi)

13、用于實(shí)現(xiàn) parable，包括面向所有基本數(shù)據(jù)類(lèi)型的封裝類(lèi)Hadoop的鍵和值所采用的數(shù)據(jù)類(lèi)型可以超出Hadoop自身所支持的基本類(lèi)型，你可以自己定義只要實(shí)現(xiàn)了它的Writable或者 parable接口Hadoop數(shù)據(jù)類(lèi)型Hadoop-MapperMapper一個(gè)類(lèi)要做為mapper,需要繼承Mapper類(lèi)，Mapper接口負(fù)責(zé)數(shù)據(jù)處理階段。它采用的形式為Mapper泛型，這里的鍵和值類(lèi)分別實(shí)現(xiàn)了 parable和Writable接口,其只有一個(gè)方法map,用于處理一個(gè)單獨(dú)的鍵/值對(duì) map(Object key, Text value, Context context）該map函數(shù)處理一個(gè)

14、給定的鍵/值對(duì)，并由context生成一個(gè)鍵/值對(duì)的列表 給reduce函數(shù)使用Hadoop提供了一些有用的mapper實(shí)現(xiàn)，部分列表如下： IdentityMapper實(shí)現(xiàn)將Mapper，將輸入直接映射到輸出 InverseMapper實(shí)現(xiàn)將Mapper，將鍵/值反轉(zhuǎn) TokenCounterMapper輸入值為分詞時(shí)，生成（token,1)對(duì)Hadoop-ReducerReducerreducer的實(shí)現(xiàn)與mapper一樣需要繼承Reducer類(lèi)，當(dāng)reducer任務(wù)接受來(lái)自各個(gè)mapper的輸出時(shí)，它按照鍵/值對(duì)中的鍵對(duì)輸入數(shù)據(jù)進(jìn)行排序，并將相同鍵的值歸并，然后調(diào)用reduce()函數(shù)迭代

15、具有相同鍵的值列表，并將結(jié)果輸出到文件reduce(Text key, Iterable values, Context context);Hadoop提供的一些基本reduce類(lèi)實(shí)現(xiàn)LongSumReducerReducerHadoop-InputFormatInputFormat接口Hadoop分割與讀取文件的方式被定義在InputFormat接口的一個(gè)實(shí)現(xiàn)中，TextInputFormat是它的一個(gè)默認(rèn)實(shí)現(xiàn)； TextInputFormat返回的鍵為每行的字節(jié)偏移量InputFormat接口的作用：1.驗(yàn)證作業(yè)的輸入是否規(guī)范2.把輸入文件切分成InputSplit3.提供RecordRe

16、ader 的實(shí)現(xiàn)類(lèi)，把InputSplit讀到Mapper中進(jìn)行處理提供了一些預(yù)定義實(shí)現(xiàn)類(lèi)： TextInputFormat作為默認(rèn)的文件輸入格式，用于讀取純文本文件，key為字節(jié)偏移量LongWritable類(lèi)型的，值為一行內(nèi)容為T(mén)ext類(lèi)型 KeyValueTextInputFormat同樣用于讀取文件，如果行被分隔符（缺省是tab）分割為兩部分，第一部分為key，剩下的部分為value；如果沒(méi)有分隔符，整行作為 key，value為空Hadoop-InputFormatInputFormat接口 CombineFileInputFormat 相對(duì)于大量的小文件來(lái)說(shuō)，hadoop更合適處理

17、少量的大文件.它是針對(duì)小文件而設(shè)計(jì)的 NLineInputformat NLineInputformat可以控制在每個(gè)split中數(shù)據(jù)的行數(shù)Hadoop-自定義輸入自定義輸入1.繼承FileInputFormat基類(lèi)2.重寫(xiě)里面的getSplits(JobContext context)方法3.重寫(xiě)createRecordReader(InputSplit split,TaskAttemptContext context)方法。Hadoop-OutputFormatOutputFormat1. TextOutputformat 默認(rèn)的輸出格式，key和value中間值用tab隔開(kāi)的。 2. SequenceFileOutputformat 將key和value以sequen