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

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