第4章分布式計算框架MapReduce本章目標(biāo)了解MapReduce計算模型了解MapReduce編程思想掌握MapReduce編程實驗源自于Google的MapReduce論文發(fā)表于2004年12月Hadoop

MapReduce是Google

MapReduce的克隆版MapReduce是一個分布式運算程序的編程框架，是用戶開發(fā)“基于hadoop的數(shù)據(jù)分析應(yīng)用”的核心框架；MapReduce核心功能是將用戶編寫的業(yè)務(wù)邏輯代碼和自帶默認(rèn)組件整合成一個完整的分布式運算程序，并發(fā)運行在一個hadoop集群上；MapReduce概述特點適用于海量數(shù)據(jù)的離線處理易于編程良好的擴(kuò)展性高容錯性MapReduce概述MapReduce采用的“分而治之”的思想，把對大規(guī)模數(shù)據(jù)集的操作，分發(fā)給一個主節(jié)點管理下的各子節(jié)點共同完成，接著通過整合各子節(jié)點的中間結(jié)果，得到最終的結(jié)果。簡單來說，MapReduce就是“分散任務(wù)，匯總結(jié)果”MapReduce概述{key1,value1}-->{key2,List<value2>}-->{key3,value3}局限性MapReduce執(zhí)行速度慢，一般在分鐘級別、小時級別后者更高與SQL相比，MapReduce過于底層不是所有的算法都能用MapReduce實現(xiàn)不適合流式計算不適合實時計算MapReduce概述MapReduce執(zhí)行數(shù)據(jù)流MapReduce基礎(chǔ)WordCount源碼解析程序位于$HADOOP_HOME/share/hadoop/mapreduce目錄下WordCount案例展示W(wǎng)ordCount作業(yè)進(jìn)行的詳細(xì)步驟1.將文件拆分為splits，由于測試用的文件較小所以每個文件作為一個split，并將文件按照行分割成<key,value>對。這一步由MapReduce框架自動完成，其中偏移量（即key值）包含了回車所占的字符數(shù)Hello

WorldBye

World分割<0,”Hello

World”><12,”Bye

World”>Hello

HadoopBye

Hadoop分割<0,”Hello

Hadoop”><13,”Bye

Hadoop”>WordCount案例展示W(wǎng)ordCount作業(yè)進(jìn)行的詳細(xì)步驟2.

將分割好的<key,value>對交給用戶定義的map方法進(jìn)行處理，生成新的<key,value>對map()<Hello,1><World,1><Bye,1><World,1>map()<Hello,1><Hadoop,1><Bye,1><Hadoop,1><0,”Hello

World”><12,”Bye

World”><0,”Hello

Hadoop”><13,”Bye

Hadoop”>WordCount案例展示W(wǎng)ordCount作業(yè)進(jìn)行的詳細(xì)步驟3.

得到map方法輸出的<key,value>對后，mapper會將他們按照key值進(jìn)行排序，并執(zhí)行Combine過程，將key值相同value值累加，得到Mapper的最終輸出結(jié)果Map端排序Map端排序<Bye,1><Hello,1><World,1><World,1><Bye,1><Hadoop,1><Hadoop,1><Hello,1><Hello,1><World,1><Bye,1><World,1><Hello,1><Hadoop,1><Bye,1><Hadoop,1><Bye,1><Hello,1><World,2><Bye,1><Hadoop,2><Hello,1>Combine過程Combine過程WordCount案例展示W(wǎng)ordCount作業(yè)進(jìn)行的詳細(xì)步驟4.得Reducer先對從Mapper接收的數(shù)據(jù)進(jìn)行排序，再交由用戶自定義的reduce方法進(jìn)行處理，得到新的<key,value>對，并作為WordCount的輸出結(jié)果。Reduce端排序<Bye,list(1,1)><Hadoop,list(2)><Hello,list(1,1)><World,list(2)><Bye,1><Hello,1><World,2><Bye,1><Hadoop,1><Hello,1><Bye,2><Hadoop,2><Hello,2><Word,2>reduce()WordCount案例展示提交MapReduce作業(yè)到Y(jié)ARN上運行hadoop

jar

<jar>

[mainClass]

args...MapReduce編程模型Mapper的實現(xiàn)Reducer的實現(xiàn)Job的配置MapReduce編程基礎(chǔ)-數(shù)據(jù)類型MapReduce編程基礎(chǔ)-數(shù)據(jù)類型org.apache.hadoop.io.Writable接口Hadoop重新實現(xiàn)序列化機(jī)制Hadoop數(shù)據(jù)類型都實現(xiàn)該接口Hadoop基本數(shù)據(jù)類型IntWritable：整型數(shù)LongWritable：長整型數(shù)FloatWritable：浮點數(shù)DoubleWritable：雙字節(jié)數(shù)BooleanWritable：布爾型ByteWritable：單字節(jié)，對應(yīng)Java中byte類型BytesWritable：字節(jié)數(shù)組，對應(yīng)Java中byte[]MapReduce編程基礎(chǔ)-數(shù)據(jù)類型Hadoop其它常用數(shù)據(jù)類型Text：使用UTF8格式存儲的文本，對String類型的重寫ObjectWritable：是一種對多類型的封裝可以用于Java的基本類型，String等如同Java的Object類NullWritable：是一個占位符，序列化長度為零當(dāng)<key,value>中的key或value為空時使用ArrayWritable：針對數(shù)組的數(shù)據(jù)類型MapWritable：對java.util.Map的實現(xiàn)InputFormat接口MapReduce編程基礎(chǔ)-輸入格式MapReduce輸入格式TextInputFormat：默認(rèn)的數(shù)據(jù)輸入格式KeyValueTextInputFormat：讀取文本鍵值對SequenceFileInputFormat：讀取二進(jìn)制鍵值對hello

hadoophello

hi

world讀取<0,

hello

hadoop><12,

hello

hi

world>hello

hadoophello

hi

world讀取<hello,hadoop><hello,

hi

world>LongWritableTextTextTextMapReduce編程基礎(chǔ)-輸入格式OutputFormat接口MapReduce編程基礎(chǔ)-輸出格式MapReduce輸出格式TextOutputFormat：默認(rèn)的數(shù)據(jù)輸出格式，以鍵值對的方式把數(shù)據(jù)寫入一個文本文件SequenceFileOutputFormat：輸出二進(jìn)制鍵值對的順序文件hellohadoop寫入<1,

hello><2,

hadoop>LongWritableTextMapReduce編程基礎(chǔ)-輸出格式MapReduce編程基礎(chǔ)實現(xiàn)Mapper繼承org.apache.hadoop.mapreduce.Mapper實現(xiàn)Reducer繼承org.apache.hadoop.mapreduce.Reducer創(chuàng)建并配置Job實例化org.apache.hadoop.mapreduce.Job]步驟一：實現(xiàn)Mapper繼承Mapper類并重寫map()方法public

class

Mapper<KEYIN,VALUEIN,KEYOUT,VALUEOUT>{protected

void

map(KEYIN

key,VALUEIN

value,Context

context){context.write((KEYOUT)key,(VALUEOUT)value);}...//省略其它方法定義}依次為：輸入鍵、輸入值、輸出鍵、輸出值的類型MapReduce任務(wù)所在的上下文，是map任務(wù)和reduce任務(wù)間的橋梁Hello

hadoop1Hello

hadoop2Hello

hadoop3split<0,Hello

hadoop1><13,Hello

hadoop2><26,Hello

hadoop3>輸入分片作為map()的輸入{public

void

map(Object

key,

Textvalue,

Context

context)IntWritable

valueOut=

new

IntWritable(1);StringTokenizer

token

=

new

StringTokenizer(value.toString());while

(token.hasMoreTokens())

{Text

keyOut=new

Text(token.nextToken());context.write(keyOut,

valueOut);}}<Hello,1><hadoop1,1><Hello,1><hadoop2,1><Hello,1><hadoop3,1>步驟一：實現(xiàn)MapperMapper類的四個方法：protected

void

setup(Context

context):執(zhí)行一次，可以在MapTask執(zhí)行開始之前進(jìn)行初始化工作，如可以打開文件等，默認(rèn)什么都不做。其中context參數(shù)記錄了

Mapper執(zhí)行的上下文環(huán)境，同時作為Mapper和Reducer執(zhí)行過程中各方法的橋梁，也就是一個可共享數(shù)據(jù)的區(qū)域。protectdvoid

map(KEYIN

key,VALUEIN

value,Context

context):循環(huán)輸入分片中的數(shù)據(jù)集，需要重寫該方法對每個鍵值對進(jìn)行處理。每個map操作只處理一個輸入分片，每個分片已經(jīng)被劃分為若干個記錄，每個記錄就是一個鍵值對，map操作一個接一個的處理記錄。其中輸入?yún)?shù)key表示每行在數(shù)據(jù)集中的字節(jié)偏移量，value為每次循環(huán)的輸入數(shù)據(jù)，為分片中的一行數(shù)據(jù)，這是由數(shù)據(jù)輸入格式?jīng)Q定的。步驟一：實現(xiàn)MapperMapper類的四個方法：

protected

void

cleanup(Context

context):執(zhí)行一次，可以在MapTask執(zhí)行之后進(jìn)行收尾工作

如關(guān)閉文件。public

void

run(Context

context):對Mapper的執(zhí)行進(jìn)行完全控制。最后執(zhí)行cleanup工作。步驟二：實現(xiàn)Reducer繼承Reducer并重寫reduce()方法public

class

Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT>{protected

void

reduce(KEYIN

key,Iterable<VALUEIN>

values,Context

context){for(VALUEIN

value

:

values){context.write((KEYOUT)key,(VALUEOUT)value);}}...//省略其它方法定義}public

void

reduce(Text

key,Iterable<IntWritable>

values,Context

context)

{intsum

=

0;for

(IntWritable

val

:

values)

{sum+=val.get();//將每個單詞的每次統(tǒng)計求和}context.write(key,

new

IntWritable(sum));}map<Hello,1><hadoop1,1><Hello,1><hadoop2,1><Hello,1><hadoop3,1><Hello,3><hadoop1,1><hadoop2,1><hadoop3,1><Hello,[1,1,1]><hadoop1,[1]><hadoop2,[1]><hadoop3,[1]>shuffle步驟二：實現(xiàn)ReducerReducer類也有四個方法：protected

void

setup(Contextcontext):執(zhí)行一次，在ReduceTask執(zhí)行開始之前進(jìn)行初始化protectdvoid

reduce(KEYINkey,Iterable<VALUE>values,Contextcontext):對每個鍵值對執(zhí)行一次，需要重寫此方法。注意第二個參數(shù)是Iterable接口，意味著他是一個可以被循環(huán)訪問的集合，但本質(zhì)上還是輸入的鍵值對中的值，在map階段和reduce階段之間還會進(jìn)行shuffle處理，目的在于對map階段的結(jié)果進(jìn)行分區(qū)(partition)，排序(sort)和溢寫(spill)。Shuffle處理過程基于鍵值對進(jìn)行排序和分組，簡單來說就是按同名鍵進(jìn)行歸類形成新的鍵值對，其鍵保持不變，而值是所有同名鍵的值得集合。protected

void

cleanup(Context

contxt):執(zhí)行一次，在ReduceTask執(zhí)行之后進(jìn)行收尾工作。public

void

run(Context

context):對Reducer的執(zhí)行進(jìn)行完全控制。最后執(zhí)行cleanup工作。步驟三：創(chuàng)建MapReduce作業(yè)實例化、配置Job類Configuration

conf=new

Configuration();//制定hdfs相關(guān)參數(shù)

Job

job=new

Job(conf,"word

count");//創(chuàng)建作業(yè)

job.setJarByClass(WordCount.class);//通過class查找job的jar文件

job.setMapperClass(TokenizerMapper.class);//指定Mapperjob.setReducerClass(IntSumReducer.class);//指定Reducerjob.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);//設(shè)置輸入文件路徑或目錄FileInputFormat.addInputPath(job,

new

Path(“hdfs://node1:9000/input/file”));//設(shè)置輸出文件目錄FileOutputFormat.setOutputPath(job,

new

Path(“hdfs://node1:9000/output/wordcount”));System.exit(job.waitForCompletion(true)?0:1);//提交作業(yè)同時指定Mapper和Reducer的鍵/值輸出類型;Mapper的鍵/值輸入類型由輸入格式?jīng)Q定；Reducer的鍵/值輸入類型由

Mapper的輸出類型決定步驟三：創(chuàng)建MapReduce作業(yè)實通過上面得知，一個job提交后會順序經(jīng)過input、split、map、shuffle、reduce這些階段，并且job一旦提交后，會進(jìn)入完全地自動化執(zhí)行過程，在這個過程中，用戶除了監(jiān)控作業(yè)的執(zhí)行情況和強(qiáng)制終止作業(yè)之外，不能進(jìn)行其他任何干預(yù)。因此在job提交之前，用戶需要所有該配置的參數(shù)配置完畢。Job對象指定作業(yè)執(zhí)行規(guī)范，可以用它來控制整個作業(yè)的運行、創(chuàng)建。配置作業(yè)的主要步驟如下：1.創(chuàng)建作業(yè)和基本配置。Configuration

conf

=

new

Configuration();

//制定hdfs相關(guān)參數(shù)Job

job

=

new

Job(conf，“word

count”);

//創(chuàng)建作業(yè)或者Job

job

=

Job.getInstance（conf，“word

count”）;job.setJarByClass(WordCount.class);

//指定jar文件Job.setNumReduceTasks(2);

//指定reduceTask數(shù)量步驟三：創(chuàng)建MapReduce作業(yè)-過程詳解需要注意的是setJarByClass方法，表示查找包含WordCount.class的jar文件。從該jar文件中應(yīng)能夠找到定義的Mapper和Reducer類，實際上執(zhí)行“hadoop

jarxxx.jar”命令時首先從“xx.jar”文件中查找，如果找不到則在JVM的classpath中繼續(xù)查找，找到后該jar文件會被上傳到集群緩存中并重命名為job.jar，仍未找到則不會上傳，到此作業(yè)提交已結(jié)束。綜合上述，最終要保證在執(zhí)行MapTask或ReduceTask時能夠找到我們定義的Mapper類和Reducer類?？梢酝ㄟ^兩種方法指定位置：a）setJarByClass(Class

class)方法，適用于Mapper、Reducer類與Job定義在統(tǒng)一jar文件中時，這是推薦用法，當(dāng)他們不在統(tǒng)一jar文件中，但如果Mapper，Reducer類所在jar文件位于當(dāng)前JVM的classpath中，也可以。b）通過setJar(String

jarFile)方法，jarFile為絕對路徑，適用于Mapper、Reducer類與Job定義在不同的jar文件中且不在當(dāng)前JVM的classpath中時。步驟三：創(chuàng)建MapReduce作業(yè)-過程詳解Job對象指定作業(yè)執(zhí)行規(guī)范，可以用它來控制整個作業(yè)的運行、創(chuàng)建。配置作業(yè)的主要步驟如下：2.設(shè)置Mapper和Reducer類型3.輸入/輸出文件格式在示例中，沒有設(shè)置輸入/輸出文件格式，前面提到Hadoop默認(rèn)為TextInputFormat和TextOutputFormat。在TextInputFormat中，每個輸入分片都會單獨作為Mapper的輸入，之后每行數(shù)據(jù)都會生產(chǎn)一條記錄并表示為<key,value>的形式，如果需要更改輸入格式，可以通過Job對象的

setinputFormatClass()方法來設(shè)置，而輸出格式使用setOutPutFormatClass()方法設(shè)置。job.setMapperClass(TokenizerMapper.class);

//指定Mapperjob.setReducerClass(IntSumReducer.class);

//指定Reducer步驟三：創(chuàng)建MapReduce作業(yè)-過程詳解Job對象指定作業(yè)執(zhí)行規(guī)范，可以用它來控制整個作業(yè)的運行、創(chuàng)建。配置作業(yè)的主要步驟如下：4.輸出類型上面兩行同時指定了Mapper和Reducer的key、value輸出類型，若Mapper和Reducer的key、value輸出類型不相同，則需要單獨指定Map的輸出類型，比如Mapper輸出<Text,Text>，而Reducer輸出<Text,IntWritable>時，需要使用如下設(shè)置：Job.setOu