大數(shù)據(jù)技術原理與操作應用 第4章 MapReduce 分布式計算系統(tǒng)

第4章MapReduce分布式計算系統(tǒng)1MapReduce運行模式MapReduce運行流程MapReduce編程案例分析使用MapReduce實現(xiàn)反向索引MapReduce基本思想MapReduce處理流程MapReduce編程模型2?學習目標理解掌握掌握掌握理解MapReduce的核心思想12掌握MapReduce的編程模型4掌握MapReduce常見編程組件的使用3掌握MapReduce的工作原理本章將對MapReduce進行系統(tǒng)的介紹,讓讀者對MapReduce有一個全面的認識。章節(jié)概要HadoopMapReduce是一個面向大規(guī)模數(shù)據(jù)集的簡單易用的并行計算模型,它極大地方便了編程人員在不會分布式并行編程的情況下,將自己的程序運行在分布式系統(tǒng)上。

4.1 MapReduce介紹4.1.1 MapReduce基本思想MapReduce是Google提出的大規(guī)模并行計算框架,應用于大規(guī)模廉價集群上的大數(shù)據(jù)并行處理。MapReduce采用“分而治之”的設計思想,將輸入的大量數(shù)據(jù)(這些數(shù)據(jù)之間不存在或有較少的依賴關系)采用一定的劃分方法進行分片,然后將一個數(shù)據(jù)分片交由一個任務去處理,這些任務并行計算,最后再匯總所有任務的處理結果。MapReduce將大數(shù)據(jù)計算任務劃分成多個子任務,然后由各個分節(jié)點并行計算,最后通過整合各個節(jié)點的中間結果,將各個子任務的結果進行合并,得到最終結果。MapReduce借助函數(shù)式編程設計思想,將大數(shù)據(jù)處理過程主要拆分為Map(映射)和Reduce(歸約)兩個模塊。Map(映射)用來將輸入的大量鍵值對映射成新的鍵值對,Reduce(歸約)負責收集整理Map操作生成的中間結果,并進行輸出。

4.1 MapReduce介紹4.1.1 MapReduce基本思想MapReduce是一個并行計算與運行軟件框架,能自動完成計算任務的并行化處理,自動劃分計算數(shù)據(jù)和計算任務,在集群節(jié)點上自動分配和執(zhí)行任務以及收集計算結果,為程序員隱藏系統(tǒng)底層細節(jié)。這樣程序員就不需要考慮數(shù)據(jù)的存儲、劃分、分發(fā)、結果收集和錯誤恢復等諸多細節(jié)問題,這些問題都交由系統(tǒng)自行處理,大大減少了軟件開發(fā)人員的負擔。

4.1 MapReduce介紹4.1.2 MapReduce處理流程MapReduce是一種并行編程模型,將計算分為兩個階段:Map階段和Reduce階段。首先將輸入數(shù)據(jù)劃分成多個塊,由多個Map任務并行計算。MapReduce對Map任務的結果進行聚集和混洗,然后提供給Reduce任務作為其輸入數(shù)據(jù)集。最終通過合并Reduce任務的輸出得到最終結果。MapReduce數(shù)據(jù)處理流程如圖4.1所示。圖4.1MapReduce數(shù)據(jù)處理流程

4.1 MapReduce介紹4.1.1 MapReduce基本思想①從存儲系統(tǒng)中讀取輸入文件內(nèi)容,存儲系統(tǒng)可以是本地文件系統(tǒng)或者HDFS文件系統(tǒng)等。對輸入文件的每一行解析成一個<key,value>對,在默認情況下,key表示行偏移量,value表示這行的內(nèi)容。②每一個<key,value>對調(diào)用一次map函數(shù)。程序員需要根據(jù)實際的業(yè)務需要重寫map()方法,對輸入的<key,value>對進行處理,轉(zhuǎn)換為新的<key,value>對輸出。(1)Map任務處理

4.1 MapReduce介紹4.1.1 MapReduce基本思想①對Map輸出的<key,value>對進行分區(qū),并將結果通過網(wǎng)絡復制到不同的Reducer節(jié)點上。②將不同分區(qū)的數(shù)據(jù)按照key進行排序,相同key的value放到一個集合中,形成新的鍵值對,即<key,list(value)>對,記為<key,VALUE>。(2)Shuffle與Sort

4.1 MapReduce介紹4.1.1 MapReduce基本思想①調(diào)用Reduce函數(shù)處理前面得到的每一個<key,VALUE>。程序員需要根據(jù)實際的業(yè)務需要重寫reduce()方法。②將Reduce函數(shù)的輸出保存到文件系統(tǒng)中。(3)Reduce任務處理

4. 2MapReduce運行機制4.2.1MapReduce編程模型簡介一個MapReduce作業(yè)通常將輸入的數(shù)據(jù)集拆分成多個獨立的塊,這些塊被Map任務以并行的方式進行計算。MapReduce將Map任務的輸出進行排序,然后將排序后的結果作為Reduce任務的輸入。作業(yè)的輸入和輸出數(shù)據(jù)均被存儲在文件系統(tǒng)中。MapReduce框架負責任務調(diào)度、監(jiān)控和重新執(zhí)行失敗的任務。HadoopMapReduce為用戶提供了五個可編程組件,分別是InputFormat、Mapper、Partitioner、Reducer和OutputFormat。還有一個組件稱為Combiner,實際上是一個局部的Reducer,通常用于MapReduce程序性能方面,不屬于必備組件。一般情況下,用戶只需要編寫Mapper和Reducer類即可,其他類已由MapReduce實現(xiàn),可以直接使用。

4. 2MapReduce運行機制4.2.2 MapReduce運行模式Hadoop的MapReduceJob的運動模式可以分為本地運行模式和集群運行模式。本地運行模式在編寫MapReduce程序時不帶集群的配置文件,也就是MapReduce程序不要配置YARN作為運行框架,配置“=local”,就可以實現(xiàn)程序的本地運行。MapReduce的本地運行模式無須啟動遠程的Hadoop集群,MapReduce程序會被提交給本地執(zhí)行器LocalJobRunner在本地以單進程的形式運行。輸入數(shù)據(jù)及輸出結果可以放在本地文件系統(tǒng),也可以放在HDFS上。本地運行模式非常便于進行業(yè)務邏輯調(diào)試,只要在Eclipse中設置斷點即可。

4. 2MapReduce運行機制4.2.2 MapReduce運行模式(2)集群運行模式首先需要啟動YARN,Job

會提交到YARN框架中去執(zhí)行,訪問“http://master:8088”可以查看Job執(zhí)行狀態(tài)。在MapReduce的集群運行模式下,MR程序?qū)峤唤oYARN集群Resourcemanager,分發(fā)到多個節(jié)點上并發(fā)執(zhí)行。輸入數(shù)據(jù)和輸出結果一般位于HDFS文件系統(tǒng)。將MapReduce程序提交集群的實現(xiàn)方法有以下幾種:①將程序打包成jar包,上傳到服務器,然后在集群的任意節(jié)點上調(diào)用Hadoop命令啟動集群執(zhí)行。②在Linux的Eclipse中直接運行main方法,將程序提交到集群中去運行,但采用此種方法,項目中要帶YARN的配置。③在Windows的Eclipse中直接運行main方法,也可以提交給集群去運行,但需要作更多的修改。

4. 2MapReduce運行機制4.2.3MapReduce運行流程

4. 2MapReduce運行機制4.2.3MapReduce運行流程(1)MapTask工作原理MapTask負責Map階段的整個數(shù)據(jù)處理流程,Map階段并行度由客戶端提交作業(yè)時的切片個數(shù)決定。①Split階段。MapReduce從文件系統(tǒng)中讀取文件后,會首先對讀取的文件進行輸入分片(inputsplit)的劃分。輸入分片存儲的并非數(shù)據(jù)本身,而是一個分片長度和一個記錄數(shù)據(jù)的位置的數(shù)組。一般情況下,以HDFS的一個塊的大小作為一個分片(也可以按split設置值來切片)。②一個輸入分片分配給一個Map任務。Map任務分配完成后,再對傳進來的分片進一步分解成一批鍵值對(<key,value>對),每一個鍵值對調(diào)用程序員事先編寫的Map函數(shù)進行邏輯處理,輸出新的<key,value>對。③數(shù)據(jù)處理完成后,會調(diào)用collect()函數(shù)進行結果的收集和輸出。collect()函數(shù)將新生成的<key,value>對進行分區(qū),并寫入一個環(huán)形內(nèi)存緩沖區(qū)中(默認為100MB)。

4. 2MapReduce運行機制4.2.3MapReduce運行流程(1)MapTask工作原理④溢寫。當環(huán)形內(nèi)存緩沖區(qū)滿后(一般為寫滿緩沖區(qū)大小的80%),MapReduce會將數(shù)據(jù)寫到本地磁盤上生成一個臨時文件。首先,對需要溢寫的緩存區(qū)內(nèi)的數(shù)據(jù)進行分區(qū),分區(qū)的個數(shù)由Reduce任務的個數(shù)決定,有多少個Reduce任務就劃分為多少個分區(qū),這樣可以保證Reduce任務分到均衡的數(shù)據(jù)。然后對每個分區(qū)中的數(shù)據(jù)進行排序,經(jīng)過排序后,數(shù)據(jù)按分區(qū)聚集在一起,且同一分區(qū)內(nèi)所有數(shù)據(jù)按照key有序。最后進行溢寫操作。如果用戶設置了Combiner,則還需要將排序后的結果以分區(qū)為單位合并成大文件,以避免同時打開大量文件和同時讀取大量小文件產(chǎn)生的隨機讀取帶來的開銷。

4. 2MapReduce運行機制4.2.3MapReduce運行流程(2)Shuffle工作原理Shuffle是MapReduce最為關鍵的一個階段,主要負責將Map端生成的數(shù)據(jù)按鍵排好序傳遞到Reduce端,包括兩個部分,即Map端的Shuffle和Reduce端的Shuffle。1)Map端的Shuffle將Map輸出的<key,value>對放到環(huán)形緩沖區(qū)中,當緩沖區(qū)空間寫滿80%時,準備將緩沖區(qū)中數(shù)據(jù)寫入到磁盤,這樣可以保證寫入到內(nèi)存緩沖區(qū)和寫入到磁盤并行進行,而不用中止Map。溢寫之前,先按照分區(qū)和key等對數(shù)據(jù)進行排序,然后按分區(qū)將數(shù)據(jù)寫入臨時文件。如果有必要,還會進行合并操作,也就是將按分區(qū)輸出的小文件合并成大文件,以提高效率。

4. 2MapReduce運行機制4.2.3MapReduce運行流程(2)Shuffle工作原理2)Reduce端的ShuffleReduce任務通過HTTP向各個Map任務復制它所需要的數(shù)據(jù),Map任務會在內(nèi)存或磁盤上對數(shù)據(jù)進行合并,Reduce任務復制過來的數(shù)據(jù)有些放在內(nèi)存中,而有些放在磁盤上。Map輸出數(shù)據(jù)已經(jīng)是有序的,Reduce會對Map傳遞過來的數(shù)據(jù)不斷地進行合并。一般Reduce端是一邊復制數(shù)據(jù)一邊合并數(shù)據(jù),復制和合并是重疊的。合并完成后,便得到了Reduce端的輸入文件。

4. 2MapReduce運行機制4.2.3MapReduce運行流程(3)ReduceTask工作原理①ReduceTask負責Reduce階段的整個數(shù)據(jù)處理流程。ReduceTask的并行度可以通過“job.setNumReduceTasks()”進行手動設置。ReduceTask默認值是“1”,輸出文件個數(shù)為1個。若ReduceTask設置為“0”,表示沒有Reduce階段,輸出文件個數(shù)與Map個數(shù)一致。②ReduceTask從各個MapTask上通過網(wǎng)絡遠程復制數(shù)據(jù),每個Map傳來的數(shù)據(jù)都是有序的。針對復制過來的數(shù)據(jù),如果數(shù)據(jù)量較小,則直接存放在內(nèi)存中;如果數(shù)據(jù)量大小超過一定閾值,則寫到磁盤上。③為了避免內(nèi)存使用過多或磁盤上存在大量小文件,在ReduceTask進行拷貝的同時,會啟動后臺線程對內(nèi)存和磁盤上的文件進行合并。因此,Reduce端一邊復制數(shù)據(jù),一邊合并數(shù)據(jù)。

4. 2MapReduce運行機制4.2.3MapReduce運行流程(3)ReduceTask工作原理④為了將key相同的數(shù)據(jù)聚集在一起,ReduceTask會對所有數(shù)據(jù)進行一次歸并排序,將相同key的value值放到一個列表中,生成形如<key,list(value)>的鍵值對。⑤對每一個鍵值對調(diào)用程序員事先編寫的Reduce()函數(shù)進行計算,并將結果輸出到文件系統(tǒng)中。

4. 2MapReduce運行機制4.2.4 MapReduce編程案例WordCount分析單詞統(tǒng)計(WordCount)是MapReduce經(jīng)典案例之一,用于統(tǒng)計大量文件中每個單詞出現(xiàn)的次數(shù),本節(jié)通過分析一個單詞統(tǒng)計案例來深入剖析MapReduce編程模型。WordCount的目標是將多個inputsplit文件輸入最終得到文件的詞頻統(tǒng)計,如圖4.3所示。圖4.3統(tǒng)計大量文件中的單詞詞頻

4. 2MapReduce運行機制4.2.4 MapReduce編程案例WordCount分析1)文件轉(zhuǎn)換圖4.4文件轉(zhuǎn)換成<key,value>對過程(1)Map階段

4. 2MapReduce運行機制4.2.4 MapReduce編程案例WordCount分析2)map()方法作用圖4.5 map()方法的轉(zhuǎn)換過程(1)Map階段

4. 2MapReduce運行機制4.2.4 MapReduce編程案例WordCount分析2)map()方法作用(1)Map階段public

static

classWordCountMap

extendsMapper<LongWritable,Text,Text,IntWritable>{

private

final

IntWritable

one=new

IntWritable(1);

privateTextword=newText();

@Override

protected

voidmap(LongWritable

key,Textvalue, Mapper<LongWritable,Text,Text,IntWritable>.Contextcontext)

throws

IOException,InterruptedException{ Stringline=value.toString();

StringTokenizer

token=new

StringTokenizer(line);

while(token.hasMoreTokens()){

word.set(token.nextToken());

context.write(word,one); } } }

4. 2MapReduce運行機制4.2.4 MapReduce編程案例WordCount分析3)Map端的Shuffle(1)Map階段圖4.6Map端的Shuffle過程

4. 2MapReduce運行機制4.2.4 MapReduce編程案例WordCount分析1)Reduce端的Shuffle(2)Reduce階段圖4.7Reduce端的Shuffle過程

4. 2MapReduce運行機制4.2.4 MapReduce編程案例WordCount分析2)reduce()方法作用(2)Reduce階段圖4.8Reduce()方法的作用過程

4. 2MapReduce運行機制4.2.4 MapReduce編程案例WordCount分析2)reduce()方法作用(2)Reduce階段public

static

class

WordCountReduce

extendsReducer<Text,IntWritable,Text,IntWritable>{

@Override

protected

voidreduce(Textkey,Iterable<IntWritable>values, Reducer<Text,IntWritable,Text,IntWritable>.Contextcontext)

throws

IOException,InterruptedException{

int

sum=0;

for(IntWritable

val:values){

sum+=val.get(); }

context.write(key,new

IntWritable(sum)); }}

4. 2MapReduce運行機制4.2.4 MapReduce編程案例WordCount分析(3)作業(yè)設置

public

static

voidmain(String[]args)

throws

IOException,ClassNotFoundException,InterruptedException{

//Configuration類代表作業(yè)的配置 Configurationconf=newConfiguration(); Jobjob=newJob(conf);

//指定主類為WordCount.class

job.setJarByClass(WordCount.class);

job.setJobName("wordcount");

//設置輸出key類型為Text，輸出值類型為IntWritable

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

//指定Mapper類和Reducer類

job.setMapperClass(WordCountMap.class);

job.setReducerClass(WordCountReduce.class);

//設置輸入輸出格式類

job.setInputFormatClass(TextInputFormat.class);

job.setOutputFormatClass(TextOutputFormat.class);

//設置輸入數(shù)據(jù)文件路徑和輸出文件存放路徑

FileInputFormat.addInputPath(job,newPath("hdfs://master:9000/wordcounttest/word.txt"));

FileOutputFormat.setOutputPath(job,newPath("hdfs://master:9000/wordcounttest/output"));

//提交作業(yè)并等待執(zhí)行完成

job.waitForCompletion(true); }

4. 3案例:使用MapReduce實現(xiàn)反向索引4.3.1Windows下Eclipse遠程開發(fā)環(huán)境搭建(1)安裝配置JDK1）下載JDK1.8，注意版本正確。2)環(huán)境變量設置①配置JDK環(huán)境變量在“計算機”上單擊鼠標右鍵,選擇“屬性”,在彈出的對話框中單擊“高級系統(tǒng)設置”,如圖4.9所示。圖4.9 高級系統(tǒng)設置

4. 3案例:使用MapReduce實現(xiàn)反向索引4.3.1Windows下Eclipse遠程開發(fā)環(huán)境搭建(1)安裝配置JDK在“系統(tǒng)屬性”對話框的“高級”選項卡中,單擊“環(huán)境變量”,如圖4.10所示。圖4.10系統(tǒng)屬性

4. 3案例:使用MapReduce實現(xiàn)反向索引4.3.1Windows下Eclipse遠程開發(fā)環(huán)境搭建(1)安裝配置JDK在“環(huán)境變量”對話框中,新建系統(tǒng)變量“JAVA_HOME”,輸入值為JDK的安裝路徑,如圖4.11所示。圖4.11新建系統(tǒng)變量“JAVA_HOME”

4. 3案例:使用MapReduce實現(xiàn)反向索引4.3.1Windows下Eclipse遠程開發(fā)環(huán)境搭建(1)安裝配置JDK編輯系統(tǒng)變量CLASSPATH,如果沒有該變量,則單擊“新建”按鈕進行創(chuàng)建。在編輯框中,輸入變量值為JDK的安裝目錄的lib目錄和jre下的lib目錄。注意,第一個英文句號表示當前目錄,不能省略。如圖4.12所示。圖4.12編輯系統(tǒng)變量CLASSPATH

4. 3案例:使用MapReduce實現(xiàn)反向索引4.3.1Windows下Eclipse遠程開發(fā)環(huán)境搭建(1)安裝配置JDK編輯系統(tǒng)變量Path,在變量值的前面添加JDK安裝目錄的bin目錄。注意,末尾的分號不能省略。如圖4.13所示。至此,JDK安裝完成。圖4.13編輯系統(tǒng)變量Path

4. 3案例:使用MapReduce實現(xiàn)反向索引4.3.1Windows下Eclipse遠程開發(fā)環(huán)境搭建(1)安裝配置JDK②測試JDK是否安裝成功JDK安裝完成后需要測試其是否安裝成功。按“Win+R”鍵彈出“運行”對話框,輸入“cmd”,在命令提示符中輸入“java-version”查看JDK版本是否正確,如圖4.14所示。圖4.14查看JDK版本

4. 3案例:使用MapReduce實現(xiàn)反向索引4.3.1Windows下Eclipse遠程開發(fā)環(huán)境搭建(1)安裝配置JDK②測試JDK是否安裝成功輸入“javac”和“java”,如果能正確輸出信息,則JDK配置成功,如圖4.15和4.16所示。如果輸出“Commandnotfound”,說明JDK配置失敗,需要重新配置。JDK的配置非常重要,一定要保證Java環(huán)境變量配置正確。圖4.16執(zhí)行java命令圖4.15執(zhí)行javac命令

4. 3案例:使用MapReduce實現(xiàn)反向索引4.3.1Windows下Eclipse遠程開發(fā)環(huán)境搭建(2)配置Eclipse在Windows上遠程開發(fā)MapReduce項目需要使用插件“hadoop-eclipse-plugin-2.7.7.jar”(2.7.7表示對應Hadoop版本號,需保持一致)。將下載好的插件放入Eclipse安裝目錄下的dropins文件夾中。以管理員身份運行解壓縮軟件將“hadoop-2.7.7.tar.gz”解壓到Windows下,由于Linux文件格式和Windows不一樣,在解壓過程中可能會有錯誤提示,可以忽略。將bin目錄下的hadoop.dll和winutils.exe(可以到網(wǎng)上自行下載)這兩個文件復制到“C:\Windows\System32\”目錄下。

4. 3案例:使用MapReduce實現(xiàn)反向索引4.3.1Windows下Eclipse遠程開發(fā)環(huán)境搭建(2)配置Eclipse重新打開Eclipse,單擊Windows→Preferences,在左側(cè)窗口中單擊“HadoopMap/Reduce”選項,在窗口右側(cè)設置Hadoop安裝路徑,如圖4.17所示。注意,Hadoop的安裝路徑既不能有空格,也不能有中文。圖4.17配置Windows下Hadoop安裝路徑

4. 3案例:使用MapReduce實現(xiàn)反向索引4.3.1Windows下Eclipse遠程開發(fā)環(huán)境搭建(2)配置Eclipse配置Map/ReduceLocations。單擊Eclipse中Window→Perspective→OpenPerspective→Other,單擊“Map/Reduce”,在下方窗口中單擊“Map/ReduceLocations”,在其下空白處右擊“NewHadoopLocation”選項。在彈出的對話框中進行Hadoop的配置,如圖4.18所示。其中“1”處配置“LocationName”,任意名稱都可以,這里輸入名稱為“master”?！?”處的Host輸入Hadoop集群中master節(jié)點的IP地址?！?”和“4”處分別設置通信端口號?！?”處設置通信的用戶名,這里設置為“apache”。圖4.18配置Hadooplocation

4. 3案例:使用MapReduce實現(xiàn)反向索引4.3.1Windows下Eclipse遠程開發(fā)環(huán)境搭建(2)配置Eclipse配置hostname的解析。為了使Windows系統(tǒng)與Hadoop集群順利通信,還需要配置hostname的解析。在Windows下打開“C:\Windows\System32\drivers\etc\hosts”文件,添加如下內(nèi)容:00master01slave102slave2

4. 3案例:使用MapReduce實現(xiàn)反向索引4.3.1Windows下Eclipse遠程開發(fā)環(huán)境搭建(2)配置Eclipse進行通信測試啟動Hadoop集群后,打開Eclipse,單擊左側(cè)的DFSLocations,如果連接成功,在ProjectExplorer的DFSLocations下會看到master下的HDFS集群中的文件,如圖4.19所示。此時,可以直接在Eclipse中操作HDFS集群上的文件。圖4.19DFSLocations展現(xiàn)

4. 2MapReduce運行機制4.3.2 使用MapReduce實現(xiàn)反向索引(1)案例說明反向索引也稱為倒排索引或反向檔案,用來存儲某個單詞在一個文檔或一組文檔中的存儲位置的映射。它是文檔檢索系統(tǒng)中最常用的數(shù)據(jù)結構,搜索引擎就是利用反向索引來進行搜索的。通過反向索引可以快速地通過文章內(nèi)的單詞反向檢索獲取包含這個單詞的文檔標識列表,從而完成巨大文件的快速搜索。

4. 2MapReduce運行機制4.3.2 使用MapReduce實現(xiàn)反向索引(1)案例說明假設有兩個文件a.txt和b.txt,其內(nèi)容如下:a.txt：helloworldhellospring

b.txt:helloapachehellohadoop通過反向索引后，得到的結果為：apacheb.txt:1hadoopb.txt:1helloa.txt:2;b.txt:2springa.txt:1worlda.txt:1

4. 2MapReduce運行機制4.3.2 使用MapReduce實現(xiàn)反向索引(1)案例說明整個實現(xiàn)過程:map函數(shù)解析輸入的文檔,輸出一系列<單詞,文檔標識及單詞次數(shù)>的鍵值對。MapReduce將鍵值對按照單詞進行排序,然后局部合并相同單詞的文檔,形成一批“<單詞,list(文檔標識及單詞次數(shù))>”。reduce函數(shù)再將這些鍵值對進行進一步地排序、合并,最后輸出反向索引結果。MapReduce反向索引實現(xiàn)原理示意圖如圖4.20所示。圖4.20MapReduce反向索引實現(xiàn)原理示意圖

4. 2MapReduce運行機制4.3.2 使用MapReduce實現(xiàn)反向索引(2)數(shù)據(jù)集本案例的數(shù)據(jù)集為“InverseData.zip”,包含三個英文文件,first.txt、second.txt和third.txt。首先,需要將這三個文件上傳到HDFS的“/mrtestlin/”目錄下。

文件上傳到DFS的