第5章MapReduce應(yīng)用開發(fā)

《大數(shù)據(jù)技術(shù)與應(yīng)用基礎(chǔ)》21世紀(jì)高等院?！霸朴?jì)算和大數(shù)據(jù)”人才培養(yǎng)規(guī)劃教材第5章MapReduce應(yīng)用開發(fā)人民郵電出版社能力CAPACITY要求了解Mapreduce的定義、功能和發(fā)展歷程。理解Map過程與Reduce過程的工作模式與任務(wù)調(diào)度。掌握使用Hadoop進(jìn)行分布式運(yùn)算的方法與技能。編寫和運(yùn)行第一個(gè)MapReduce程序之前的準(zhǔn)備配置HadoopMapReduce開發(fā)環(huán)境MapReduce應(yīng)用案例一、配置HadoopMapReduce開發(fā)環(huán)境Windows764位操作系統(tǒng)hadoop2.6.0Eclipsehadoop-eclipse-plugin-2.6.0.jar系統(tǒng)環(huán)境及所需文件安裝Eclipse安裝配置時(shí)所用的Eclipse版本是EclipseIDEforJavaEEDevelopers，其下載地址如下：/downloads/download.php?file=/technology/epp/downloads/release/neon/R/eclipse-jee-neon-R-win32-x86_64.zip&mirror_id=448一、配置HadoopMapReduce開發(fā)環(huán)境安裝Eclipse將下載好的Eclipse壓縮包解壓，即完成安裝配置。向Eclipse中添加插件插件包hadoop-eclipse-plugin-2.6.0.jar充當(dāng)?shù)淖饔檬鞘笶clipse連接Hadoop集群，方便在Eclipse上面進(jìn)行hadoopmapreduce方面的開發(fā)。該插件包可以在網(wǎng)上搜索關(guān)鍵字進(jìn)行下載，將下載下來的插件包復(fù)制進(jìn)Eclipse安裝目錄中的plugins文件夾中，再重啟Eclipse即可。在Eclipse中添加map/reduceLocations窗口的方式：依次選擇“Window”→“ShowView”→“Other”，打開MapReduceTools選擇Map/ReduceLocations然后單擊“OK”按鈕。這樣就能在Eclipse上通過Map/ReduceLocations窗口進(jìn)行Hadooplocation添加刪除和配置等操作。一、配置HadoopMapReduce開發(fā)環(huán)境通過Eclipse來管理HadoopHDFS修改集群中namenode節(jié)點(diǎn)的hdfs-site.xml，添加如下內(nèi)容:<property><name>dfs.permissions</name><value>false</value></property>在Map/ReduceLocations窗口中添加一個(gè)名為hadoop的Hadooplocation然后重啟Hadoop集群，即可在Eclipse中管理HDFS，進(jìn)行一些增刪改查看等操作。配置HadoopMapReduce開發(fā)環(huán)境編寫和運(yùn)行第一個(gè)MapReduce程序之前的準(zhǔn)備MapReduce應(yīng)用案例一、編寫和運(yùn)行第一個(gè)MapReduce程序之前的準(zhǔn)備Windows764位操作系統(tǒng)hadoop2.6.0Eclipsehadoop.dll&winutils.exejavajdk1.8.0系統(tǒng)環(huán)境及所需文件建立運(yùn)行MapReduce程序的依賴（1）下載安裝Hadoop（2）下載hadoop.dll&winutils.exe并將其添加到windows下的Hadoop安裝目錄中的bin目錄里。（3）在Eclipse中設(shè)置Windows中Hadoop安裝目錄的絕對(duì)路徑?！癢indow”→“Preferences”→“HadoopMap/Reduce”。一、編寫和運(yùn)行第一個(gè)MapReduce程序之前的準(zhǔn)備在Eclipse中新建一個(gè)maven工程；建立編寫MapReduce程序的依賴包（1）在EclipseProjectExplorer窗口右擊，選擇“New”→“Other”；（2）選擇“MavenProject”→“Next”；（3）選擇相應(yīng)的maven架包→“Next”；（4）根據(jù)實(shí)際情況進(jìn)行配置→“Finish”；在新建的maven工程里，找到pom.xml。在這里可以添加編寫MapReduce程序所需要的依賴包。進(jìn)入/，在搜索欄上查找你所需要添加的依賴包，選擇對(duì)應(yīng)的Hadoop版本號(hào)，然后單擊進(jìn)入。將其中Maven選項(xiàng)欄里的內(nèi)容復(fù)制粘貼到pom.xml中保存，即可自動(dòng)下載。一、編寫和運(yùn)行第一個(gè)MapReduce程序之前的準(zhǔn)備下載完成后的結(jié)果建立編寫MapReduce程序的依賴包編寫和運(yùn)行第一個(gè)MapReduce程序之前的準(zhǔn)備MapReduce應(yīng)用案例配置HadoopMapReduce開發(fā)環(huán)境三、MapReduce應(yīng)用案例MapReduce應(yīng)用案例數(shù)據(jù)去重單表關(guān)聯(lián)單詞計(jì)數(shù)多表關(guān)聯(lián)排序三、MapReduce應(yīng)用案例在HDFS根目錄下有一文件words，其內(nèi)容為：helloahell0bhelloc單詞計(jì)數(shù)主要其程序代碼如下：publicstaticclassMyReducerextendsReducer<Text,IntWritable,Text,IntWritable>{

privateIntWritablesum=newIntWritable();

protectedvoidreduce(Textkey,Iterable<IntWritable>values,

Reducer<Text,IntWritable,Text,IntWritable>.Contextcontent)

throwsIOException,InterruptedException{ Integercount=0; for(IntWritablevalue:values){ count+=value.get(); } sum.set(count); content.write(key,sum); }

}三、MapReduce應(yīng)用案例單詞計(jì)數(shù)主要其程序代碼如下：publicstaticvoidmain(String[]args)throwsException{

if(args.length<2){ args=newString[]{

"hdfs://master:9000/words",

"hdfs://master:9000/out" }; }對(duì)代碼執(zhí)行RunonHadoop操作，執(zhí)行結(jié)果如右圖三、MapReduce應(yīng)用案例數(shù)據(jù)去重兩個(gè)文本文件如下：現(xiàn)通過使用MapReduce去除這兩個(gè)文件中重復(fù)的部分。主要程序代碼如下//實(shí)現(xiàn)map函數(shù)publicvoidmap(Objectkey,Textvalue,Contextcontext)

throwsIOException,InterruptedException{

line=value;

context.write(line,newText(""));}//reduce將輸入中的key復(fù)制到輸出數(shù)據(jù)的key上，并直接輸出

publicstaticclassReduceextendsReducer<Text,Text,Text,Text>{

//實(shí)現(xiàn)reduce函數(shù)

publicvoidreduce(Textkey,Iterable<Text>values,Contextcontext)

throwsIOException,InterruptedException{

context.write(key,newText(""));

}

}

三、MapReduce應(yīng)用案例數(shù)據(jù)去重對(duì)代碼執(zhí)行RunonHadoop操作，執(zhí)行結(jié)果如下圖三、MapReduce應(yīng)用案例排序兩個(gè)文本文件如下：現(xiàn)通過使用MapReduce去除這兩個(gè)文件中重復(fù)的部分。主要程序代碼如下//實(shí)現(xiàn)map函數(shù)

publicvoidmap(Objectkey,Textvalue,Contextcontext)

throwsIOException,InterruptedException{

Stringline=value.toString();

data.set(Integer.parseInt(line));

context.write(data,newIntWritable(1));

}//reduce將輸入中的key復(fù)制到輸出數(shù)據(jù)的key上，

//然后根據(jù)輸入的value-list中元素的個(gè)數(shù)決定key的輸出次數(shù)

//用全局linenum來代表key的位次

publicstaticclassReduceextendsReducer<IntWritable,IntWritable,IntWritable,IntWritable>{privatestaticIntWritablelinenum=newIntWritable(1);三、MapReduce應(yīng)用案例排序//實(shí)現(xiàn)reduce函數(shù)

publicvoidreduce(IntWritablekey,Iterable<IntWritable>values,Contextcontext)

throwsIOException,InterruptedException{

for(IntWritableval:values){

context.write(linenum,key);

linenum=newIntWritable(linenum.get()+1);}}}publicstaticvoidmain(String[]args)throwsException{

Configurationconf=newConfiguration();

//這句話很關(guān)鍵

Jobjob=Job.getInstance(conf,"DataSort");

job.setJarByClass(DataSort.class);對(duì)代碼執(zhí)行RunonHadoop操作，執(zhí)行結(jié)果如下圖三、MapReduce應(yīng)用案例單表關(guān)聯(lián)在HDFS根目錄下family_in文件目錄中創(chuàng)建包含5-1所示的child_parent（孩子父母）表中內(nèi)容的文件，要求通過MapReduce輸出如表5-2所示的grandchild_grandparent（孫子爺奶）表中的內(nèi)容。三、MapReduce應(yīng)用案例單表關(guān)聯(lián)主要其程序代碼如下：publicstaticclassMapextendsMapper<Object,Text,Text,Text>{//實(shí)現(xiàn)map函數(shù)publicvoidmap(Objectkey,Textvalue,Contextcontext)throwsIOException,InterruptedException{Stringchildname=newString();//孩子名稱Stringparentname=newString();//父母名稱Stringrelationtype=newString();//左右表標(biāo)識(shí)

//輸入的一行預(yù)處理文本StringTokenizeritr=newStringTokenizer(value.toString());String[]values=newString[2];inti=0;

while(itr.hasMoreTokens()){

values[i]=itr.nextToken();i++;}

if(values[0].compareTo("child")!=0){

childname=values[0];parentname=values[1];//輸出左表relationtype="1";context.write(newText(values[1]),newText(relationtype+

"+"+childname+"+"+parentname));//輸出右表relationtype="2";context.write(newText(values[0]),newText(relationtype+"+"+childname+"+"+parentname));}

}接下頁三、MapReduce應(yīng)用案例單表關(guān)聯(lián)主要其程序代碼如下：

publicstaticclassReduceextendsReducer<Text,Text,Text,Text>{//實(shí)現(xiàn)reduce函數(shù)

publicvoidreduce(Textkey,Iterable<Text>values,Contextcontext)throwsIOException,InterruptedException{//輸出表頭

if(0==time){context.write(newText("grandchild"),newText("grandparent"));

time++;}intgrandchildnum=0;String[]grandchild=newString[10];intgrandparentnum=0;String[]grandparent=newString[10];

Iteratorite=values.iterator();while(ite.hasNext()){

Stringrecord=ite.next().toString(); 接下頁

三、MapReduce應(yīng)用案例單表關(guān)聯(lián)intlen=record.length();inti=2;if(0==len){

continue;}

//取得左右表標(biāo)識(shí)charrelationtype=record.charAt(0);//定義孩子和父母變量Stringchildname=newString();Stringparentname=newString();

//獲取value-list中value的childwhile(record.charAt(i)!='+'){

childname+=record.charAt(i);i++;}i=i+1;//獲取value-list中value的parent

while(i<len){

parentname+=record.charAt(i);i++;}

//左表，取出child放入grandchildren

if('1'==relationtype){grandchild[grandchildnum]=childname;grandchildnum++;}