第4章 分布式計(jì)算框架MapReduce

第4章分布式計(jì)算框架MapReduce4.1認(rèn)識MapReduce嵌入式系統(tǒng)基本概念為了降低軟件開發(fā)人員的編程難度，MapReduce框架隱藏了很多內(nèi)部功能的實(shí)現(xiàn)細(xì)節(jié)，實(shí)現(xiàn)了自動并行處理。軟件開發(fā)人員可以在不會分布式并行編程的情況下，將自己的程序運(yùn)行在分布式系統(tǒng)上，完成大數(shù)據(jù)集的計(jì)算。4.1.1分布式并行編程嵌入式系統(tǒng)基本概念單個計(jì)算機(jī)的CPU、內(nèi)存和硬盤等資源是有限的，單個計(jì)算機(jī)在處理海量數(shù)據(jù)時，無法高效地完成大量運(yùn)算。一種有效的解決方案是借助于分布式并行編程來提高程序性能。分布式程序運(yùn)行在大規(guī)模計(jì)算機(jī)集群上，可以并行執(zhí)行大規(guī)模數(shù)據(jù)處理任務(wù)，從而獲得海量的計(jì)算能力。同時通過向集群中增加新的計(jì)算機(jī)，可以很容易地?cái)U(kuò)充集群的計(jì)算能力。MapReduce最早是由Google提出的一種面向大規(guī)模數(shù)據(jù)處理的分布式并行計(jì)算模型，HadoopMapReduce是它的開源實(shí)現(xiàn)。4.1.2MapReduce簡介嵌入式系統(tǒng)基本概念用MapReduce來處理的數(shù)據(jù)集必須具備這樣的特點(diǎn)：待處理的數(shù)據(jù)集可以分解成許多小的數(shù)據(jù)集，而且每一個小數(shù)據(jù)集都可以完全獨(dú)立地進(jìn)行并行處理。MapReduce采用“分而治之”的核心思想，即將一個大數(shù)據(jù)集通過一定的數(shù)據(jù)劃分方法，分成多個較小的數(shù)據(jù)集，各小的數(shù)據(jù)集之間不存在依賴關(guān)系，將這些小的數(shù)據(jù)集分配給不同的節(jié)點(diǎn)去處理，最后將處理的結(jié)果進(jìn)行匯總，從而形成最終的結(jié)果。4.1.2MapReduce簡介嵌入式系統(tǒng)基本概念

MapReduce設(shè)計(jì)的一個理念就是“計(jì)算向數(shù)據(jù)靠攏”，因?yàn)樵诖笠?guī)模數(shù)據(jù)環(huán)境下，移動數(shù)據(jù)需要大量的網(wǎng)絡(luò)傳輸開銷。只要有可能，MapReduce就會將Map程序分配到數(shù)據(jù)所在的節(jié)點(diǎn)上運(yùn)行，從而減少節(jié)點(diǎn)之間的數(shù)據(jù)傳輸開銷。最簡單的MapReduce應(yīng)用程序至少包含3個部分：一個map函數(shù)、一個reduce函數(shù)和一個main函數(shù)。在運(yùn)行一個MapReduce程序時，整個處理過程被分為Map階段和Reduce階段，每個階段都是用鍵值對（key/value）作為輸入和輸出。main函數(shù)是MapReduce應(yīng)用程序的入口，它將文件輸入/輸出和作業(yè)控制結(jié)合起來。4.1.3MapReduce的運(yùn)行環(huán)境嵌入式系統(tǒng)基本概念

YARN（YetAnotherResourceNegotiator，另一種資源協(xié)調(diào)者）是Hadoop2.0中的資源管理和調(diào)度框架，YARN的目標(biāo)就是實(shí)現(xiàn)“一個集群多個框架”，即在一個集群上部署一個統(tǒng)一的資源調(diào)度管理框架YARN，在YARN之上不但可以運(yùn)行MapReduce，而且還可以運(yùn)行Spark、Storm、Tez等計(jì)算框架。YARN的引入為集群在資源統(tǒng)一管理和數(shù)據(jù)共享等方面帶來了很大方便。4.1.3MapReduce的運(yùn)行環(huán)境嵌入式系統(tǒng)基本概念

YARN采用主從架構(gòu)（Master/Slave），包括：ResourceManager、ApplicationMaster和NodeManager三個核心組件。ResourceManager運(yùn)行在主節(jié)點(diǎn)，負(fù)責(zé)整個集群的資源管理和分配。每個應(yīng)用程序擁有一個ApplicationMaster，ApplicationMaster管理一個在YARN內(nèi)運(yùn)行的應(yīng)用程序?qū)嵗?，?fù)責(zé)申請資源、任務(wù)調(diào)度和任務(wù)監(jiān)控；NodeManager運(yùn)行在從節(jié)點(diǎn)，整個集群有多個NodeManager，負(fù)責(zé)單節(jié)點(diǎn)資源的管理和使用。4.1.4Hadoop內(nèi)置數(shù)據(jù)類型嵌入式系統(tǒng)基本概念1.ByteWritable：字節(jié)類型。2.IntWritable：整型類型。3.LongWritable：長整型類型。4.BooleanWritable：布爾類型。5.FloatWritable：單精度浮點(diǎn)數(shù)類型。6.DoubleWritable：雙精度浮點(diǎn)數(shù)類型。7.Text：使用UTF8格式存儲的文本類型。8.NullWritable：空對象，當(dāng)<key,value>中的key或value為空時使用。4.1.4Hadoop內(nèi)置數(shù)據(jù)類型嵌入式系統(tǒng)基本概念【例4-1】常見Hadoop數(shù)據(jù)類型的應(yīng)用importorg.apache.hadoop.io.DoubleWritable;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.io.Text;4.1.4Hadoop內(nèi)置數(shù)據(jù)類型嵌入式系統(tǒng)基本概念publicclassHadoopDataTypeTest{ /*使用hadoop的Text類型*/ publicstaticvoidtestText(){ System.out.println("testText:"); Texttext=newText("Helloworld!"); System.out.println(text.toString()); System.out.println(text.find("r")); System.out.println(text.getLength()); text.set("Hello,hadoop!"); System.out.println(text.toString()); }4.1.4Hadoop內(nèi)置數(shù)據(jù)類型嵌入式系統(tǒng)基本概念/*使用hadoop的IntWritable類型*/ publicstaticvoidtestIntWritable(){ System.out.println("testIntWritable:"); IntWritableintWritable=newIntWritable(5); System.out.println("intWritable="+intWritable); inti=intWritable.get()+1; intWritable.set(i); System.out.println("intWritable="+intWritable); }4.1.4Hadoop內(nèi)置數(shù)據(jù)類型嵌入式系統(tǒng)基本概念/*使用hadoop的DoubleWritable類型*/ publicstaticvoidtestDoubleWritable(){ System.out.println("testDoubleWritable:"); DoubleWritabledoubleWritable=newDoubleWritable(6.5); System.out.println("doubleWritable="+doubleWritable); doubled=doubleWritable.get()+1.3; doubleWritable.set(d); System.out.println("doubleWritable="+doubleWritable); }4.1.4Hadoop內(nèi)置數(shù)據(jù)類型嵌入式系統(tǒng)基本概念publicstaticvoidmain(Stringargs[]){ testText(); testIntWritable(); testDoubleWritable(); }}4.1.4Hadoop內(nèi)置數(shù)據(jù)類型嵌入式系統(tǒng)基本概念圖4-1程序運(yùn)行結(jié)果4.2.1MapReduce工作流程概述嵌入式系統(tǒng)基本概念圖4-2MapReduce的工作流程4.2.2Shuffle過程分析嵌入式系統(tǒng)基本概念圖4-3MapReduce的Shuffle過程4.3MapReduce入門示例：WordCount嵌入式系統(tǒng)基本概念本節(jié)以Hadoop自帶的WordCount程序?yàn)槔?，分析WordCount的設(shè)計(jì)思路、介紹WordCount編程方法和運(yùn)行步驟。4.3.1WordCount程序任務(wù)嵌入式系統(tǒng)基本概念

表4-1WordCount程序任務(wù)4.3.1WordCount程序任務(wù)嵌入式系統(tǒng)基本概念

表4-2WordCount的輸入和輸出實(shí)例4.3.2準(zhǔn)備被統(tǒng)計(jì)的文件嵌入式系統(tǒng)基本概念創(chuàng)建f1.txt和f2.txt輸入如下命令，創(chuàng)建test2目錄，上傳f1.txt和f2.txthadoopfs-mkdir/test2//在HDFS根目錄下創(chuàng)建test2hadoopfs-putf1.txt/test2//上傳f1.txt到HDFS的/test2下hadoopfs-putf2.txt/test2//上傳f2.txt到HDFS的/test2下4.3.3WordCount的執(zhí)行過程分析嵌入式系統(tǒng)基本概念圖4-5拆分輸入數(shù)據(jù)4.3.3WordCount的執(zhí)行過程分析嵌入式系統(tǒng)基本概念圖4-6執(zhí)行用戶定義的map()4.3.3WordCount的執(zhí)行過程分析嵌入式系統(tǒng)基本概念圖4-7使用Combine函數(shù)的Map端Shuffle過程4.3.3WordCount的執(zhí)行過程分析嵌入式系統(tǒng)基本概念圖4-8Reduce端的操作4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念importjava.io.IOException;importjava.util.StringTokenizer;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.io.Text;4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念importorg.apache.hadoop.mapreduce.Job;importorg.apache.hadoop.mapreduce.Mapper;importorg.apache.hadoop.mapreduce.Reducer;importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;importorg.apache.hadoop.util.GenericOptionsParser;4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念publicclassWordCount{/**自定義的TokenizerMapper類繼承Mapper類。<Object,Text,Text,IntWritable>分別用來指定Map的輸入key值類型、輸入value值類型、輸出key值類型、輸出value值類型*/publicstaticclassTokenizerMapperextendsMapper<Object,Text,Text,IntWritable>{privatefinalstaticIntWritableone=newIntWritable(1);privateTextword=newText();4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念/**key記錄的數(shù)據(jù)的偏移量，value是每次split提供給程序讀取的一行數(shù)據(jù)*/publicvoidmap(Objectkey,Textvalue,Contextcontext)throwsIOException,InterruptedException{/**StringTokenizer用于字符串分解，默認(rèn)的分隔符是空格("")、制表符(\t)、換行符(\n)、回車符(\r)*/StringTokenizeritr=newStringTokenizer(value.toString());while(itr.hasMoreTokens()){word.set(itr.nextToken());context.write(word,one);//以<word,1>的形式輸出

}}}4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念

/**自定義的IntSumReducer類繼承Reducer類。<Text,IntWritable,Text,IntWritable>分別用來指定Reduce的輸入key值類型、輸入value值類型、輸出key值類型、輸出value值類型*/publicstaticclassIntSumReducerextendsReducer<Text,IntWritable,Text,IntWritable>{privateIntWritableresult=newIntWritable();4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念/**reduce將輸入的key值作為輸出的key值，將key值對應(yīng)的list各元素值加起來，作為value值輸出*/publicvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{intsum=0;4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念for(IntWritableval:values){sum+=val.get();//將相同的單詞對應(yīng)的值加一起

}result.set(sum);context.write(key,result);}}4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念/**main()的輸入?yún)?shù)決定了輸入數(shù)據(jù)的文件位置，以及輸出數(shù)據(jù)存儲的位置；創(chuàng)建一個Configuration對象時，會獲取Hadoop的配置信息；通過Job的對象可設(shè)置Hadoop程序運(yùn)行時的環(huán)境變量*/publicstaticvoidmain(String[]args)throwsException{Configurationconf=newConfiguration();String[]otherArgs=newGenericOptionsParser(conf,args).getRemainingArgs();4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念if(otherArgs.length<2){System.err.println("Usage:wordcount<in>[<in>...]<out>");System.exit(2);}Jobjob=Job.getInstance(conf,"wordcount");//實(shí)例化jobjob.setJarByClass(WordCount.class);//設(shè)置整個程序的類名

job.setMapperClass(TokenizerMapper.class);//為job設(shè)置Map類

job.setCombinerClass(IntSumReducer.class);//為job設(shè)置Combine類4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念job.setReducerClass(IntSumReducer.class);//為job設(shè)置Reduce類

job.setOutputKeyClass(Text.class);//為job的輸出數(shù)據(jù)設(shè)置Key類job.setOutputValueClass(IntWritable.class);//為job輸出設(shè)置value類/**為job設(shè)置輸入路徑*/for(inti=0;i<otherArgs.length-1;++i){FileInputFormat.addInputPath(job,newPath(otherArgs[i]));}4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念

/**為job設(shè)置輸出路徑*/FileOutputFormat.setOutputPath(job,newPath(otherArgs[otherArgs.length-1]));System.exit(job.waitForCompletion(true)?0:1);}}4.4.1求平均值嵌入式系統(tǒng)基本概念求平均值是數(shù)據(jù)分析中的常用操作，下面以求學(xué)生的平均成績?yōu)槔榻B使用MapReduce求平均值的方法?！纠?-2】現(xiàn)有某班學(xué)生English、Math、Chinese三門課程的成績，每門課程的成績記錄在一個文本文件中，請編寫MapReduce程序求每位學(xué)生的平均成績。4.4.1求平均值嵌入式系統(tǒng)基本概念importjava.io.IOException;importjava.util.StringTokenizer;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Job;4.4.1求平均值嵌入式系統(tǒng)基本概念importorg.apache.hadoop.mapreduce.Mapper;importorg.apache.hadoop.mapreduce.Reducer;importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;importorg.apache.hadoop.util.GenericOptionsParser;4.4.1求平均值嵌入式系統(tǒng)基本概念publicclassWordCount{

publicstaticclassTokenizerMapperextendsMapper<Object,Text,Text,IntWritable>{/**去掉finalstatic，將one修改為實(shí)例變量*/

privateIntWritableone=newIntWritable(1);privateTextword=newText();publicvoidmap(Objectkey,Textvalue,Contextcontext)throwsIOException,InterruptedException{4.4.1求平均值嵌入式系統(tǒng)基本概念

/**將輸入的數(shù)據(jù)首先按行進(jìn)行分割*/

StringTokenizeritr=newStringTokenizer(value.toString(),"\n");/**分別對每行進(jìn)行處理*/while(itr.hasMoreTokens()){/**將每行按默認(rèn)分隔符進(jìn)行分割*/

StringTokenizerlie=newStringTokenizer(itr.nextToken());/**獲取學(xué)生姓名*/

StringstrName=lie.nextToken();4.4.1求平均值嵌入式系統(tǒng)基本概念

/**獲取學(xué)生成績*/

StringstrScore=lie.nextToken();

word.set(strName);one.set(Integer.parseInt(strScore));/**輸出姓名和成績*/context.write(word,one);}}}4.4.1求平均值嵌入式系統(tǒng)基本概念publicstaticclassIntSumReducerextendsReducer<Text,IntWritable,Text,IntWritable>{privateIntWritableresult=newIntWritable();

publicvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{4.4.1求平均值嵌入式系統(tǒng)基本概念intsum=0;

intcount=0;for(IntWritableval:values){/**計(jì)算總分*/sum+=val.get();/**統(tǒng)計(jì)科目數(shù)*/

count++;}4.4.1求平均值嵌入式系統(tǒng)基本概念

/**計(jì)算平均值，并給result賦值*/

result.set((int)(sum/count));context.write(key,result);}}4.4.1求平均值嵌入式系統(tǒng)基本概念publicstaticvoidmain(String[]args)throwsException{Configurationconf=newConfiguration();String[]otherArgs=newGenericOptionsParser(conf,args).getRemainingArgs();if(otherArgs.length<2){System.err.println("Usage:wordcount<in>[<in>...]<out>");System.exit(2);}4.4.1求平均值嵌入式系統(tǒng)基本概念jobjob=Job.getInstance(conf,"wordcount");job.setJarByClass(WordCount.class);job.setMapperClass(TokenizerMapper.class);job.setCombinerClass(IntSumReducer.class);job.setReducerClass(IntSumReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);4.4.1求平均值嵌入式系統(tǒng)基本概念

for(inti=0;i<otherArgs.length-1;++i){FileInputFormat.addInputPath(job,newPath(otherArgs[i]));}FileOutputFormat.setOutputPath(job,newPath(otherArgs[otherArgs.length-1]));System.exit(job.waitForCompletion(true)?0:1);}}4.4.2簡單查詢功能的實(shí)現(xiàn)嵌入式系統(tǒng)基本概念在下面的例子中，我們借助MapReduce模擬SQL語句，實(shí)現(xiàn)簡單的查詢功能，能夠更好幫助讀者理解第6章將要學(xué)習(xí)的Hive?！纠?-3】employees.txt中保存某公司的員工信息，每位員工的信息占一行，每行數(shù)據(jù)包含員工編號（id）、姓名（name）、年齡（age）、月薪（salary）和部門編號（depts）五項(xiàng)信息。如果把employees.txt看作一張表，表名為employees。請完成查詢功能：“select*fromemployeeswhereage<30”，即查詢年齡少于30歲的員工信息。4.4.2簡單查詢功能的實(shí)現(xiàn)嵌入式系統(tǒng)基本概念//

Employees.javaimportjava.io.DataInput;importjava.io.DataOutput;importjava.io.IOException;importorg.apache.hadoop.io.Writable;4.4.2簡單查詢功能的實(shí)現(xiàn)嵌入式系統(tǒng)基本概念publicclassEmployeesimplementsWritable{privateintid;//員工編號

privateStringname;//姓名

privateintage;//年齡

privatedoublesalary;//月薪

privateintdepts;//部門編號4.4.2簡單查詢功能的實(shí)現(xiàn)嵌入式系統(tǒng)基本概念//實(shí)現(xiàn)反序列化@OverridepublicvoidreadFields(DataInputin)throwsIOException{ id=in.readInt(); name=in.readUTF(); age=in.readInt(); salary=in.readDouble(); depts=in.readInt(); }4.4.2簡單查詢功能的實(shí)現(xiàn)嵌入式系統(tǒng)基本概念

//實(shí)現(xiàn)序列化

@Override publicvoidwrite(DataOutputout)throwsIOException{ out.writeInt(id); out.writeUTF(name); out.writeInt(age); out.writeDouble(salary); out.writeInt(depts); }4.4.2簡單查詢功能的實(shí)現(xiàn)嵌入式系統(tǒng)基本概念//各種getter和setter publicintgetId(){ returnid; } publicvoidsetId(intid){ this.id=id; } publicStringgetName(){ returnname; }4.4.2簡單查詢功能的實(shí)現(xiàn)嵌入式系統(tǒng)基本概念

publicvoidsetName(Stringname){ =name; } publicintgetAge(){ returnage; } publicvoidsetAge(intage){ this.age=age; } publicdoublegetSalary(){ returnsalary; }4.4.2簡單查詢功能的實(shí)現(xiàn)嵌入式系統(tǒng)基本概念publicvoidsetSalary(doublesalary){ this.salary=salary; } publicintgetDepts(){ returndepts; } publicvoidsetDepts(intdepts){ this.depts=depts; }}4.4.2簡單查詢功能的實(shí)現(xiàn)嵌入式系統(tǒng)基本概念//

WordCount.javaimportjava.io.IOException;importjava.util.StringTokenizer;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.io.Text;4.4.2簡單查詢功能的實(shí)現(xiàn)嵌入式系統(tǒng)基本概念importorg.apache.hadoop.mapreduce.Job;importorg.apache.hadoop.mapreduce.Mapper;importorg.apache.hadoop.mapreduce.Reducer;importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;importorg.apache.hadoop.util.GenericOptionsParser;4.4.2簡單查詢功能的實(shí)現(xiàn)嵌入式系統(tǒng)基本概念publicclassWordCount{publicstaticclassTokenizerMapperextendsMapper<Object,Text,IntWritable,Employees>{/**創(chuàng)建一個IntWritable對象，作為輸出的key*/privateIntWritableid=newIntWritable();/**創(chuàng)建一個Employees對象，作為輸出的value*/privateEmployeesemployees=newEmployees();4.4.2簡單查詢功能的實(shí)現(xiàn)嵌入式系統(tǒng)基本概念publicvoidmap(Objectkey,Textvalue,Contextcontext)throwsIOException,InterruptedException{/**將文本分割成行*/StringTokenizeritr=newStringTokenizer(value.toString(),"\n");while(itr.hasMoreTokens()){/**將每行以逗號為分隔符進(jìn)行分割，分割成各字段*/ StringTokenizerlie=newStringTokenizer(itr.nextToken(),",");/**將分割后的各字段值保存到employees對象中*/ employees.setId(Integer.valueOf(lie.nextToken()));//轉(zhuǎn)換成整數(shù)4.4.2簡單查詢功能的實(shí)現(xiàn)嵌入式系統(tǒng)基本概念

employees.setName(lie.nextToken()); employees.setAge(Integer.parseInt(lie.nextToken())); employees.setSalary(Double.valueOf(lie.nextToken()));//轉(zhuǎn)換成浮點(diǎn)數(shù)

employees.setDepts(Integer.parseInt(lie.nextToken())); if(employees.getAge()<30){ id.set(employees.getId()); context.write(id,employees); }}}}4.4.2簡單查詢功能的實(shí)現(xiàn)嵌入式系統(tǒng)基本概念publicstaticclassIntSumReducerextendsReducer<IntWritable,Employees,IntWr