2015E-mail:主頁：

第七章MapReduce

（PPT版本號：2015年6月第1.0版）

《大數據技術原理與應用》溫馨提示：編輯幻燈片母版，可以修改每頁PPT的廈大校徽和底部文字提綱7.1 概述7.2 MapReduce工作流程7.3 實例分析：WordCount7.4 MapReduce的具體應用7.5 MapReduce編程實踐歡迎訪問《大數據技術原理與應用》教材官方網站：本PPT是如下教材的配套講義：21世紀高等教育計算機規劃教材《大數據技術原理與應用——概念、存儲、處理、分析與應用》（2015年6月第1版）廈門大學林子雨編著，人民郵電出版社ISBN:978-7-115-39287-97.1 概述7.1.1 分布式并行編程7.1.2 MapReduce模型簡介7.1.3 Map和Reduce函數7.1.1 分布式并行編程“摩爾定律”，大約每隔18個月性能翻一番從2005年開始摩爾定律逐漸失效，人們開始借助于分布式并行編程來提高程序性能分布式程序運行在大規模計算機集群上，集群中包括大量廉價服務器，可以并行執行大規模數據處理任務，從而獲得海量的計算能力谷歌公司最先提出了分布式并行編程模型MapReduce，HadoopMapReduce是它的開源實現

7.1.2 MapReduce模型簡介MapReduce將復雜的、運行于大規模集群上的并行計算過程高度地抽象到了兩個函數：Map和Reduce在MapReduce中，一個存儲在分布式文件系統中的大規模數據集，會被切分成許多獨立的小數據塊，這些小數據塊可以被多個Map任務并行處理MapReduce框架會為每個Map任務輸入一個數據子集，Map任務生成的結果會繼續作為Reduce任務的輸入，最終由Reduce任務輸出最后結果，并寫入到分布式文件系統中MapReduce設計的一個理念就是“計算向數據靠攏”，而不是“數據向計算靠攏”，因為，移動數據需要大量的網絡傳輸開銷MapReduce框架采用了Master/Slave架構，包括一個Master和若干個Slave。Master上運行JobTracker，Slave上運行TaskTrackerHadoop框架是用Java實現的，但是，MapReduce應用程序則不一定要用Java來寫

7.1.3 Map和Reduce函數函數輸入輸出說明Map<k1,v1>List(<k2,v2>)1.將小數據集進一步解析成一批<key,value>對，輸入Map函數中進行處理2.每一個輸入的<k1,v1>會輸出一批<k2,v2>。<k2,v2>是計算的中間結果Reduce<k2,List(v2)><k3,v3>輸入的中間結果<k2,List(v2)>中的List(v2)表示是一批屬于同一個k2的value表7-1Map和Reduce7.2 MapReduce工作流程7.2.1 工作流程概述7.2.2 MapReduce各個執行階段7.2.3 Shuffle過程詳解7.2.1 工作流程概述圖7-1MapReduce工作流程7.2.2 MapReduce各個執行階段圖7-2MapReduce工作流程中的各個執行階段7.2.3 Shuffle過程詳解圖7-3Shuffle過程

1.Shuffle過程簡介7.2.3 Shuffle過程詳解2.Map端的Shuffle過程圖7-4Map端的Shuffle過程7.2.3 Shuffle過程詳解3.Reduce端的Shuffle過程圖7-5Reduce端的Shuffle過程7.3 實例分析：WordCount7.3.1 WordCount程序任務7.3.2 WordCount設計思路7.3.3 MapReduce具體執行過程7.3.4 一個WordCount執行過程的實例7.3.1 WordCount程序任務表7-2WordCount程序任務程序WordCount輸入一個包含大量單詞的文本文件輸出文件中每個單詞及其出現次數（頻數），并按照單詞字母順序排序，每個單詞和其頻數占一行，單詞和頻數之間有間隔表7-3一個WordCount的輸入和輸出實例輸入輸出HelloWorldHelloHadoopHelloMapReduceHadoop1Hello3MapReduce1World17.3.2 WordCount設計思路首先，需要檢查WordCount程序任務是否可以采用MapReduce來實現其次，確定MapReduce程序的設計思路最后，確定MapReduce程序的執行過程7.3.3 MapReduce具體執行過程圖7-6WordCount執行過程7.3.4 一個WordCount執行過程的實例圖7-7Map過程示意圖7.3.4 一個WordCount執行過程的實例圖7-8用戶沒有定義Combiner時的Reduce過程示意圖7.3.4 一個WordCount執行過程的實例圖7-9用戶有定義Combiner時的Reduce過程示意圖7.4MapReduce的具體應用MapReduce可以很好地應用于各種計算問題，這里以關系代數運算、分組與聚合運算、矩陣-向量乘法、矩陣乘法為例，介紹如何采用MapReduce計算模型來實現各種運算7.4.1 MapReduce在關系代數運算中的應用7.4.2 分組與聚合運算7.4.3 矩陣-向量乘法7.4.4 矩陣乘法7.4.1MapReduce在關系代數運算中的應用針對數據的很多運算可以很容易地采用數據庫查詢語言來表示，即使這些查詢本身并不在數據庫管理系統中執行關系數據庫中的關系(relation)可以看做是由一系列屬性值組成的表，關系中的行稱為元組(tuple)，屬性的集合稱為關系的模式下面介紹基于MapReduce模型的關系上的標準運算，包括選擇、投影、并、交、差以及自然連接7.4.1MapReduce在關系代數運算中的應用對于關系的選擇運算，只需要Map過程就能實現，對于關系R中的每個元組t，檢測是否是滿足條件的元組，如果滿足條件，則輸出鍵值對<t,t>，也就是說，鍵和值都是t。這時的Reduce函數就只是一個恒等式，對輸入不做任何變換就直接輸出1.關系的選擇運算7.4.1MapReduce在關系代數運算中的應用假設對關系R投影后的屬性集為S，在Map函數中，對于R中的每個元組t，剔除t中不屬于S的字段得到元組，輸出鍵值對<,>。對于Map任務產生的每個鍵，可能存在一個或多個鍵值對<,>，因此，需要通過Reduce函數來剔除冗余，把屬性值完全相同的元組合并起來得到<,<,,...>>，剔除冗余后只輸出一個<,>。2.關系的投影運算7.4.1MapReduce在關系代數運算中的應用對兩個關系求并集時，Map任務將兩個關系的元組轉換成鍵值對<t,t>，Reduce任務則相當于一個剔除冗余數據的過程(合并到一個文件中)對兩個關系求交集時，使用與求并集相同的Map過程，在Reduce過程中，如果鍵t有兩個相同值與它關聯，則輸出一個元組<t,t>，如果與鍵關聯的只有一個值，則輸出空值（NULL）對兩個關系求差時，Map過程產生的鍵值對，不僅要記錄元組的信息，還要記錄該元組來自于哪個關系（R或S），Reduce過程中按鍵值相同的t合并后，與鍵t相關聯的值如果只有R（說明該元組只屬于R，不屬于S），就輸出元組，其他情況均輸出空值3.關系的并、交、差運算7.4.1MapReduce在關系代數運算中的應用有關系R(A，B)和S(B,C)，將屬性B的值分別作為它們的鍵和元組中其他屬性值關聯起來使用Map過程，把來自R的每個元組<a,b>轉換成一個鍵值對<b,<R,a>>，其中的鍵就是屬性B的值。注意，這里把關系R包含到值中，這樣做使得我們可以在Reduce階段，只把那些來自R的元組和來自S的元組進行匹配。類似地，使用Map過程，把來自S的每個元組<b,c>，轉換成一個鍵值對<b,<S,c>>所有具有相同B值的元組被發送到同一個Reduce進程中，Reduce進程負責對這些元組進行合并。假設使用k個Reduce進程，這里選擇一個哈希函數h，它可以把屬性B的值映射到k個哈希桶，每個哈希值對應一個Reduce進程。每個Map進程把鍵是b的鍵值對都發送到與哈希值h(b)對應的Reduce進程Reduce進程的輸出則是連接后的元組<a,b,c>，輸出被寫到一個單獨的輸出文件中4.關系的自然連接7.4.1MapReduce在關系代數運算中的應用以某工廠接到的訂單與倉庫貨存為例，演示關系自然連接運算的MapReduce過程4.關系的自然連接7.4.2分組與聚合運算詞頻計算就是典型的分組聚合運算在Map過程中，選擇關系的某一字段（也可以是某些屬性構成的屬性表）的值作為鍵，其他字段的值作為與鍵相關聯的值。在Reduce過程中，對鍵值相同的鍵值對的值施加某種聚合運算，如SUM(求和)、COUNT(計數)、AVG(求平均值)、MIN和MAX(求最小最大值)等，輸出則為<鍵，聚合運算結果>7.4.3矩陣-向量乘法假定一個n維向量V，其第j個元素記為，和一個的矩陣M，其第i行第j列元素記為，矩陣M和向量V的乘積是一個n維向量X，其第i個元素。矩陣M和向量V各自會在分布式文件系統（比如HDFS）中存成一個文件。假定我們可以獲得矩陣元素的行列下標，例如從矩陣元素在文件中的位置來獲得，或者從元素顯式存儲的三元組<i,j,>中來獲得。7.4.3矩陣-向量乘法每個Map任務將整個向量V和矩陣M的一個文件塊作為輸入。對每個矩陣元素，Map任務會產生鍵值對<i,>。計算所得的n個求和項的鍵都相同，即都是i。Reduce任務將所有與給定鍵i關聯的值相加即可得到

<i,>。7.4.3矩陣-向量乘法如果n的值過大，使向量V無法完全放入內存中，那么，在計算過程中就需要多次將向量的一部分導入內存，這就會導致大量的磁盤訪問一種替代方案是，將矩陣分割成多個寬度相等的垂直條，同時，將向量分割成同樣數目的水平條，每個水平條的高度等于矩陣垂直條的寬度矩陣第i個垂直條只和第i個水平條相乘。因此，可以將矩陣的每個條存成一個文件，同樣，將向量的每個條存成一個文件。矩陣某個條的一個文件塊及對應的完整向量條輸送到每個Map任務。然后，Map任務和Reduce任務可以按照前述過程進行7.4.4矩陣乘法矩陣M第i行第j列的元素記為，矩陣N中的第j行第k列的元素記為，矩陣，第i行第k列元素為。把矩陣看作一個帶有三個屬性的關系：行下標、列下標和值。因此，矩陣M可以看作關系M(I,J,V)，元組為<i，j，>，矩陣N可以看作關系N(J,K,W)，元組為<j,k,>。矩陣乘法可以看作是一個自然連接運算再加上分組聚合運算。關系M和N根據公共屬性J將每個元組連接得到元組<i,j,k,v,w>，這個五字段元組代表了兩個矩陣的元素對<,>，對矩陣元素進行求積運算后可以得到四字段元組<i,j,k,>，然后可以進行分組聚合運算，其中，I、K是分組屬性，的和是聚合結果。綜上所述，矩陣乘法可以通過兩個MapReduce運算的串聯來實現。7.4.4矩陣乘法Map函數：對每個矩陣元素產生一個鍵值對<j,<M,i,>>，對每個矩陣元素產生一個鍵值對<j,<N,k,>>Reduce函數：對每個相同鍵j，輸出所有滿足形式

<j,<i,k,>>的元組。1.自然連接階段7.4.4矩陣乘法Map函數：對自然連接階段產生的鍵值對

<j,<<>,<>,...<>>>（其中每個是對應的和的乘積），Map任務會產生p個鍵值對<<<>,>,<<>,>...,<<>,>>。Reduce函數：對每個鍵<i,k>，計算與此鍵關聯的所有值的和，結果記為<<i,k>,v>，其中，v就是矩陣P的第i行第k列的值。2.分組聚合階段7.5 MapReduce編程實踐7.5.1 任務要求7.5.2 編寫Map處理邏輯7.5.3 編寫Reduce處理邏輯7.5.4 編寫main方法7.5.5 編譯打包代碼以及運行程序7.5.1任務要求文件A的內容如下：ChinaismymotherlandIloveChina文件B的內容如下：IamfromChina期望結果如右側所示：I2is1China3my1love1am1from1motherland17.5.2編寫Map處理邏輯Map輸入類型為<key,value>期望的Map輸出類型為<單詞，出現次數>Map輸入類型最終確定為<Object,Text>Map輸出類型最終確定為<Text,IntWritable>7.5.3編寫Reduce處理邏輯在Reduce處理數據之前，Map的結果首先通過Shuffle階段進行整理Reduce階段的任務：對輸入數字序列進行求和Reduce的輸入數據為<key,Iterable容器>7.5.4編寫main方法publicstaticvoidmain(String[]args)throwsException{Configurationconf=newConfiguration();//程序運行時參數

String[]otherArgs=newGenericOptionsParser(conf,args).getRemainingArgs();if(otherArgs.length!=2){System.err.println("Usage:wordcount<in><out>");System.exit(2);}Jobjob=newJob(conf,"wordcount");//設置環境參數

job.setJarByClass(WordCount.class);//設置整個程序的類名

job.setMapperClass(MyMapper.class);//添加MyMapper類

job.setReducerClass(MyReducer.class);//添加MyReducer類

job.setOutputKeyClass(Text.class);//設置輸出類型

job.setOutputValueClass(IntWritable.class);//設置輸出類型

(job,newPath(otherArgs[0]));//設置輸入文件

(job,newPath(otherArgs[1]));//設置輸出文件

System.exit(job.waitForCompletion(true)?0:1);}7.5.5編譯打包代碼以及運行程序包功能org.apache.hadoop.conf定義了系統參數的配置文件處理方法org.apache.hadoop.fs定義了抽象的文件系統APIorg.apache.hadoop.dfsHadoop分布式文件系統（HDFS）模塊的實現org.apache.hadoop.mapredHadoop分布式計算框架MapReduce的實現，包括任務的分發調度等org.apache.hadoop.ipc網絡服務端和客戶端的工具，封裝了網絡異步I/O的基礎模塊org.apache.hadoop.io定義了通用的I/OAPI，用于針對于網絡、數據庫、文件等數據對象進行讀寫操作等7.5.5編譯打包代碼以及運行程序實驗步驟：使用java編譯程序，生成.class文件將.cla