大數(shù)據課件1第5章-MapReduce分布式計算框架

1第5章MapReduce分布式計算框架5.1MapReduce概述5.2MapReduce工作原理5.3MapReduce編程類5.4MapReduce經典案例5.1MapReduce編程模型概述5.1.1并發(fā)、并行與分布式編程的概念5.1.2MapReduce并行編程模型概述5.1.3Map函數(shù)和Reduce函數(shù)5.1.1并發(fā)、并行與分布式編程的概念并發(fā)是兩個任務可以在重疊的時間段內啟動，運行和完成，并發(fā)是獨立執(zhí)行過程的組合。有時并發(fā)確實能在同一時間間隔內完成更多的任務，也就是有利于增加任務的吞吐量。因為在單CPU的情況下，并不是所有的任務在進行的每時每刻都使用CPU，也許還要使用IO等設備。并行是任務在同一時間運行；并行是同時執(zhí)行（可能相關的）計算。并行要比并發(fā)更難得，對于單CPU來說，除了指令級別的并行，其他情況并不能實行精確的并行。5.1.1并發(fā)、并行與分布式編程的概念分布式編程的主要特征是分布和通信。采用分布式編程方法設計程序時，一個程序由若干個可獨立執(zhí)行的程序模塊組成。這些程序模塊分布于一個分布式計算機系統(tǒng)的幾臺計算機上同時執(zhí)行。分布在各臺計算機上的程序模塊是相互關聯(lián)的，它們在執(zhí)行中需要交換數(shù)據，即通信。只有通過通信，各程序模塊才能協(xié)調地完成一個共同的計算任務。分布式編程語言，往往可以由一種程序設計語言增加分布和通信的功能而構成。5.1MapReduce編程模型概述7.1.1并發(fā)、并行與分布式編程的概念7.1.2MapReduce并行編程模型概述7.1.3Map函數(shù)和Reduce函數(shù)5.1.2MapReduce并行編程模型概述MapReduce用于大規(guī)模數(shù)據集（大于1TB）的并行運算，將待求解的復雜問題，分解成等價的規(guī)模較小的若干部分，然后逐個解決，分別求得各部分的結果，把各部分的結果組成整個問題的結果。MapReduce將并行計算過程高度的抽象為Map（映射）和Reduce（規(guī)約）兩個計算過程，分別對應一個函數(shù)，這兩個函數(shù)由應用程序開發(fā)者負責具體實現(xiàn)，開發(fā)者不需要處理并行編程中的其它各種復雜問題，如分布式存儲、工作調度、負載均衡、容錯處理、網絡通信等，這些問題全部由MapReduce框架負責處理。適合用MapReduce來處理的數(shù)據集（或任務）有一個基本要求：待處理的數(shù)據集可以分解成許多小的數(shù)據集，而且每一個小數(shù)據集都可以完全并行地進行處理。5.1.2MapReduce并行編程模型概述在MapReduce中，一個存儲在分布式文件系統(tǒng)中的大規(guī)模數(shù)據集會被切分成許多獨立的小數(shù)據塊，這些小數(shù)據塊被分別提交給多個Map任務并行處理，Map任務處理后所所生成的結果作為多個Reduce任務的輸入，由Reduce任務處理生成最終結果并將其寫入分布式文件系統(tǒng)。MapReduce設計的一個理念是“計算向數(shù)據靠攏”，而不是“數(shù)據向計算靠攏”。只要有可能，MapReduce框架就會將Map程序就近地在HDFS數(shù)據所在的節(jié)點上運行，即將計算節(jié)點和存儲節(jié)點合并為一個節(jié)點，從而減少節(jié)點間的數(shù)據移動開銷。5.1MapReduce編程模型概述5.1.1并發(fā)、并行與分布式編程的概念5.1.2MapReduce并行編程模型概述5.1.3Map函數(shù)和Reduce函數(shù)5.1.3Map函數(shù)和Reduce函數(shù)MapReduce的Map函數(shù)和Reduce函數(shù)的核心思想源自于函數(shù)式編程的map和reduce函數(shù)。在函數(shù)式編程中，map函數(shù)的功能是接受一個列表list以及一個函數(shù)，將這個函數(shù)作用于這個列表中的所有成員，并返回所得結果。在Python語言中，map函數(shù)使用舉例如下：>>>L=[1,2,3,4,5]#創(chuàng)建一個列表>>>list(map((lambdax:x+5),L))#將L中的每個元素加5[6,7,8,9,10]5.1.3Map函數(shù)和Reduce函數(shù)reduce函數(shù)的功能則是接收一個列表、一個初始值以及一個函數(shù)，將該函數(shù)作為特定的組合方式，將其遞歸地應用于列表的所有成員，并返回最終結果。在Python語言中，reduce函數(shù)使用舉例如下：（1）不帶初始參數(shù)initializer的reduce()函數(shù)reduce(function,sequence)，先將sequence的第一個元素作為function函數(shù)的第一個參數(shù)和sequence的第二個元素作為function函數(shù)第二個參數(shù)進行function函數(shù)運算，然后將得到的返回結果作為下一次function函數(shù)的第一個參數(shù)和序列sequence的第三個元素作為function的第二個參數(shù)進行function函數(shù)運算，得到的結果再與第四個數(shù)據用function進行函數(shù)運算，依次進行下去直到sequence中的所有元素都得到處理。>>>defadd(x,y):#定義一個求和的函數(shù)，函數(shù)名為add returnx+y>>>reduce(add,[1,2,3,4,5])#計算列表和：1+2+3+4+5155.1.3Map函數(shù)和Reduce函數(shù)（2）帶初始參數(shù)initializer的reduce()函數(shù)reduce(function,sequence,initializer)，先將初始參數(shù)initializer的值作為function函數(shù)的第一個參數(shù)和sequence的第一個元素作為function的第二個參數(shù)進行function函數(shù)運算，然后將得到的返回結果的作為下一次function函數(shù)的第一個參數(shù)和序列sequence的第二個元素作為function的第二個參數(shù)進行function函數(shù)運算，得到的結果再與第3個數(shù)據用function進行函數(shù)運算，依次進行下去直到sequence中的所有元素都得到處理。>>>reduce(add,[2,3,4,5,6],1)#帶初始參數(shù)1，計算1+2+3+4+5+621Hadoop的MapReduce模型的Map函數(shù)和Reduce函數(shù)在函數(shù)式編程的map和reduce函數(shù)基礎上進行了細微的擴展，Map函數(shù)和Reduce函數(shù)不接收數(shù)值（如int、string類型的值），而接收鍵值對<key,value>，同時這些函數(shù)的每一個輸出也都是一個鍵值對<key,value>。12第5章MapReduce分布式計算框架5.1MapReduce概述5.2MapReduce工作原理5.3MapReduce編程類5.4MapReduce經典案例5.2MapReduce工作原理5.2.1MapReduce體系架構MapReduce是一個編程模型，用戶定義一個Map函數(shù)來處理一批鍵-值對<key,value>以生成一批中間的鍵-值對<key,value>，再定義一個Reduce函數(shù)將所有這些中間的鍵-值對<key,value>中key相同的value合并起來。MapReduce的執(zhí)行流程：將一個大數(shù)據劃分多個較小的具有同樣計算過程的數(shù)據塊將每一個數(shù)據塊分給不同的Map任務去處理，生成以鍵-值對<key,value>形式表示的許多中間結果將中間結果劃分成和Reduce任務數(shù)相等的多個分區(qū)，分別分發(fā)給不同的Reduce任務處理，具有相同key的<key,value>會被發(fā)送到同一個Reduce任務那里，Reduce任務對中間結果進行匯總計算得到最終結果。5.2MapReduce工作原理注意：不同的Map任務之間、不同的Reduce任務之間不會進行通信，用戶不能顯示地從一個計算節(jié)點向另一個計算節(jié)點發(fā)送消息，所有的數(shù)據交換都是通過MapReduce框架自身去實現(xiàn)的。MapReduce執(zhí)行流程5.2.1MapReduce體系架構5.2MapReduce工作原理MapTask作為MapReduce工作流程的前半部分，它主要經歷了6個階段。5.2.2MapTask工作原理5.2MapReduce工作原理MapTask作為MapReduce工作流程的前半部分，它主要經歷了6個階段：5.2.2MapTask工作原理（1）把輸入文件按照一定的標準切分為邏輯上的多個輸入片（InputSplit），InputSplit是MapReduce對文件進行處理和運算的輸入單位，只是一個邏輯概念，每個InputSplit并沒有對文件進行實際切割，只是記錄了要處理的數(shù)據的位置和長度。（2）把輸入片中的記錄按照一定的規(guī)則解析成“鍵-值”對，默認規(guī)則是把每一行文本內容解析成一個“鍵-值”對<key1,value1>，“鍵”是每一行的起始位置，“值”是本行的文本內容。（3）對第（2）階段中解析出來的每一個鍵值對<key1,value1>調用map方法一次，輸出零個或者多個鍵值對<key2,value2>。以每個單詞為鍵key2、以1（詞頻數(shù)）作為鍵key2對應的值value2。5.2MapReduce工作原理MapTask作為MapReduce工作流程的前半部分，它主要經歷了6個階段：5.2.2MapTask工作原理（4）為了讓Reduce可以并行處理Map的結果，需要對Map輸出的鍵值對按照一定的規(guī)則進行分區(qū)（partition）。分區(qū)是基于鍵進行的，比如鍵表示省份（如河北、河南、山東等），那么就可以按照不同省份進行分區(qū)，同一個省份的鍵值對劃分到一個分區(qū)中，默認是只有一個分區(qū)。（5）對每個分區(qū)中的鍵-值對進行排序。首先，按照鍵進行排序，對于鍵相同的鍵值對，按照值進行排序，比如三個鍵值對<2,2>、<1,3>、<2,1>，那么排序后的結果是<1,3>、<2,1>、<2,2>。如果有第六階段，那么進入第六階段；如果沒有，直接輸出到本地磁盤上。（6）對每個分區(qū)中的數(shù)據進行歸約處理，也就是reduce()處理。鍵相等的鍵值對會調用一次reduce()方法，得到<key,value-list>形式的中間結果。5.2MapReduce工作原理5.2.3ReduceTask工作原理ReduceTask的執(zhí)行過程主要經歷了4個階段，分別是Copy階段、Merge階段、Sort階段和Reduce階段。5.2MapReduce工作原理5.2.3ReduceTask工作原理（1）Copy階段：ReduceTask會主動從MapTask復制其輸出的鍵值對，如果其大小超過一定閾值，則寫到磁盤上，否則直接放到內存中。（2）Merge階段：在遠程拷貝數(shù)據的同時，ReduceTask啟動了兩個后臺線程對內存和磁盤上的文件進行合并，以防止內存使用過多或磁盤上文件過多。（3）Sort階段：用戶自定義reduce函數(shù)，其接收的輸入數(shù)據是按key進行聚集的一組數(shù)據。Hadoop采用了基于排序的策略將key相同的數(shù)據聚在一起。由于各個MapTask已經實現(xiàn)對自己的處理結果進行了局部排序，因此，ReduceTask只需對所有數(shù)據進行一次歸并排序即可。歸并排序是指將兩個或兩個以上的有序序列合并成一個有序序列。5.2MapReduce工作原理5.2.3ReduceTask工作原理（4）Reduce階段：對排序后的鍵值對<key,value-list>調用reduce函數(shù)，鍵相等的鍵值對調用一次reduce函數(shù)，每次調用會產生零個或者多個鍵值對，最后把這些輸出的鍵值對寫入到HDFS文件中。在對MapTask、ReduceTask的分析過程中，會看到很多階段都出現(xiàn)了鍵值對，容易混淆，下面對鍵值對進行編號，如圖所示。21第5章MapReduce分布式計算框架5.1MapReduce概述5.2MapReduce工作原理5.3MapReduce編程類5.4MapReduce經典案例5.3MapReduce編程類5.3.1InputFormat輸入格式類InputFormat是一個抽象類，位于org.apache.hadoop.mapreduce.InputFormat<K,V>。主要用于驗證作業(yè)數(shù)據的輸入形式和格式；將輸入數(shù)據分割為若干個邏輯意義上的InputSplit，其中每一個InputSplit將單獨作為一個Mapper基類的輸入；提供一個RecordReader類，用于將InputSplit解析成一個個<key,value>對。如果要編寫一個MapReduce程序，需要借助MapReduce提供的一些編程類（組件）來實現(xiàn)FileInputFormat的常用子類有：TextInputFormat、CombineFileInputFormat、KeyValueTextInputFormat、NLlineInputFormat、SequenceFileInputFormat等。5.3MapReduce編程類5.3.1InputFormat輸入格式類1.FileInputFormat類FileInputFormat默認按照文件長度和切片大小進行切片，切片大小默認等于block大??；切片時不考慮數(shù)據集整體，而是逐個針對每一個文件單獨切片。比如待處理數(shù)據有兩個文件：file1.txt，其大小為320MB；file2.txt，其大小為10MB。經過FileInputFormat的切片機制運算后，形成的切片信息如下：file1.txt.split1：0~128MBfile1.txt.split2：128~256MBfile1.txt.split3：256~320MBfile2.txt.split1：0~10MB5.3MapReduce編程類5.3.1InputFormat輸入格式類2.TextInputFormat類TextInputFormat是默認的InputFormat。每條記錄是一行輸入，鍵是該行在整個文件中的起始字節(jié)偏移量，值是這行的內容，不包括任何行終止符（換行符和回車符）。以下是一個示例，比如，一個分片包含了如下4條文本記錄。HelloHadoopHelloHDFSHelloSparkHelloScala每條記錄表示為以下“鍵-值”對：(0,HelloHadoop)(13,HelloHDFS)(24,HelloSpark)(36,HelloScala)5.3MapReduce編程類3.KeyValueTextInputFormat類按照指定分隔符的形式，按行分割字符串，每一行均為一條記錄，被分隔符分割為key和value。如果一行當中存在多個指定分隔符，只有第一個有效?？梢酝ㄟ^在驅動類中進行如下設置：conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR,"");來設定分隔符。默認分隔符是tab（\t）。以下是一個示例，輸入是一個包含4條記錄的分片。其中——>表示一個（水平方向的）制表符。line1——>HelloHadoopline2——>HelloHDFSline3——>HelloSparkline4——>HelloScala每條記錄表示為以下鍵/值對：(line1,HelloHadoop)(line2,HelloHDFS)(line3,HelloSpark)(line4,HelloScala)5.3MapReduce編程類4.NlineInputFormat類如果使用NlineInputFormat，代表每個map進程處理的InputSplit不再按block塊去劃分，而是按NlineInputFormat指定的行數(shù)N來劃分。即輸入文件的總行數(shù)/N=切片數(shù)，如果不整除，切片數(shù)=商+1。以下是一個示例，以4行輸入為例。HelloHadoopHelloHDFSHelloSparkHelloScala例如，如果N是2，共分成兩個分片，每個輸入分片包含兩行。分片1的每條記錄表示為以下鍵/值對(0,HelloHadoop)(13,HelloHDFS)這里的鍵和值與TextInputFormat生成的一樣。5.3MapReduce編程類5.3.2Mapper基類簡單來說，Map是一些單個任務，Mapper類就是實現(xiàn)Map任務的類。Hadoop提供了一個抽象的Mapper基類，該基類提供了一個map方法，默認情況下，Mapper基類中的map方法是沒有做任何處理的。如果想自定義map方法，只需繼承Mapper基類并重寫map方法即可。Mapper基類的map()方法定義如下：protectedvoidmap(KEYINkey,VALUEINvalue,Contextcontext)throwsIOException,InterruptedException{context.write((KEYOUT)key,(VALUEOUT)value);}

其中，輸入參數(shù)key是傳入map的鍵；value是對應鍵的值；context是環(huán)境對象參數(shù)，保存了作業(yè)運行的上下文信息，例如作業(yè)配置信息、InputSplit信息、任務ID。5.3MapReduce編程類5.3.2Mapper基類Hadoop使用MapReduce框架為每個由作業(yè)的InputFormat產生的InputSplit生成一個Map任務。下面編寫一個簡單的文檔詞頻統(tǒng)計Mapper類：importjava.io.IOException;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Mapper;/***本類繼承自Mapper類*Mapper<LongWritable,Text,Text,IntWritable>的參數(shù)含義為*LongWritable表示文本偏移量，Text表示讀取的一行文本*Text表示map函數(shù)輸出的key的類型，IntWritable表示map函數(shù)輸出的value的類型*/5.3MapReduce編程類5.3.2Mapper基類publicstaticclassWordCountMapperextendsMapper<LongWritable,Text,Text,IntWritable>{//定義一個靜態(tài)常量one并將它的值初始化為1privatefinalstaticIntWritableone=newIntWritable(1);//定義一個靜態(tài)Text類的引用wordprivateTextword=newText();//完成詞頻統(tǒng)計的map方法

publicvoidmap(LongWritablekey,Textvalue,Mapper<LongWritable,Text,Text,IntWritable>.Contextcontext)throwsIOException,InterruptedException{StringTokenizeritr=newStringTokenizer(value.toString());while(itr.hasMoreTokens()){word.set(itr.nextToken());context.write(word,one);}}}5.3MapReduce編程類5.3.3Combiner合并類每一個map都可能會產生大量的本地輸出，Combiner的作用就是對map端的輸出先做一次合并，以減少在map和reduce節(jié)點之間的數(shù)據傳輸量，以提高網絡IO性能，其具體的作用如下所述。（1）Combiner可以看做局部的Reducer。（2）Combiner作用是合并相同的key對應的value。（3）在Mapper階段，不管Combiner被調用多少次，都不應改變Reduce的輸出結果。（4）Combiner通常與Reducer的邏輯是一樣的，一般情況下不需要單獨編寫Combiner，直接使用Reducer的實現(xiàn)就可以了。（5）Combiner在Job中是通過job.setCombinerClass(Reducer.class)來設置的。并不是所有的場景都可以使用Combiner，如適合于Sum()求和，并不適合Average()求平均數(shù)。5.3MapReduce編程類5.3.4Partitioner分區(qū)類Partitioner處于Mapper階段，當Mapper處理好數(shù)據后，這些數(shù)據需要經過Partitioner進行分區(qū)，來選擇不同的Reducer處理，從而將Mapper的輸出結果均勻的分布在Reducer上面執(zhí)行。對于map輸出的每一個鍵值對，系統(tǒng)都會給定一個partition，partition值默認通過計算key的hash值后對Reducetask的數(shù)量取模獲得。如果一個鍵值對的partition值為1，意味著這個鍵值對會交給第一個Reducer處理。用戶自定義Partitioner，需要繼承Partitioner類，實現(xiàn)它提供的一個方法。5.3MapReduce編程類5.3.5Sort排序類Sort是Map過程所產生的中間數(shù)據在送給Reduce進行處理之前所要經過的一個過程。當map()函數(shù)處理完輸入數(shù)據之后，會將中間數(shù)據存在本地的一個或者幾個文件中，并且針對這些文件內部的記錄進行一次升序的快速排序。然后，在Map任務將所有的中間數(shù)據寫入本地文件并進行快速排序之后，系統(tǒng)會對這些排好序的文件做一次歸并排序，并將排好序的結果輸出到一個大的文件中。5.3MapReduce編程類5.3.6Reducer類Map輸出的中間鍵值對集合[(k2,v2)]經過合并處理后，把鍵相同的鍵值對的值合并到一個列表里得到中間結果(k2,[v2])。Reduce對(k2,[v2])進行處理，并產生最終的某種形式的結果輸出[(k3,v3)]。即Reducer根據key將中間數(shù)據集合處理合并為更小的數(shù)據結果集。5.3.7輸出格式類OutputFormat在MapReduce框架中，OutputFormat抽象類負責把Reducer處理完成的Key-Value寫出到本地磁盤或HDFS上，默認所有計算結果會以part-r-00000的命名方式輸出成多個文件，并且輸出的文件數(shù)量與Reduce數(shù)量一致。00000是關聯(lián)到某個Reduce任務的分區(qū)的ID號。MapReduce提供多種輸出格式，用戶可以靈活設置輸出的路徑、文件名、輸出格式等。OutputFormat常見的輸出格式實現(xiàn)類還包括TextOutputFormat、SequenceFileOutputFormat和DBOutputFormat。34第5章MapReduce分布式計算框架5.1MapReduce概述5.2MapReduce工作原理5.3MapReduce編程類5.4MapReduce經典案例5.4MapReduce經典案例5.4.1WordCount執(zhí)行流程示例WordCount是Hadoop自帶的示例程序之一，其功能是統(tǒng)計輸入文件（也可以是輸入文件夾內的多個文件）中每個單詞出現(xiàn)的次數(shù)。WordCount的基本設計思路是分別統(tǒng)計每個文件中單詞出現(xiàn)的次數(shù)，然后累加不同文件中同一個單詞出現(xiàn)次數(shù)。WordCount執(zhí)行流程包括以下幾個階段：（1）將文件拆分成split，測試用到的兩個文件內容如圖5-6所示：5.4MapReduce經典案例5.4.1WordCount執(zhí)行流程示例WordCount執(zhí)行流程包括以下幾個階段：（2）將分割好的<key1,value1>對交給用戶定義的map方法進行處理，每個Map任務中，以每個單詞為鍵key2、以1（詞頻數(shù)）作為鍵key2對應的值value2生成新的鍵值對<key2,value2>，然后輸出，如圖5-8所示：5.4MapReduce經典案例5.4.1WordCount執(zhí)行流程示例WordCount執(zhí)行流程包括以下幾個階段：（3）得到map方法輸出的<key2,value2>對后，Mapper會將它們按照key值進行排序，并執(zhí)行Combine過程，將key值相同的value值累加，得到Mapper的最終輸出結果，如圖5-9所示：5.4MapReduce經典案例5.4.1WordCount執(zhí)行流程示例WordCount執(zhí)行流程包括以下幾個階段：（4）Reducer先對從Mapper接收的數(shù)據進行排序，再交由用戶自定義的reduce()方法進行處理，將相同主鍵下的所有值相加，得到新的<key3,value3>對作為最終的輸出結果，如圖