第5章MapReduce应用开发

《大数据技术与应用基础》21世纪高等院校“云计算和大数据”人才培养规划教材第5章MapReduce应用开发人民邮电出版社能力CAPACITY要求了解Mapreduce的定义、功能和发展历程。理解Map过程与Reduce过程的工作模式与任务调度。掌握使用Hadoop进行分布式运算的方法与技能。编写和运行第一个MapReduce程序之前的准备配置HadoopMapReduce开发环境MapReduce应用案例一、配置HadoopMapReduce开发环境Windows764位操作系统hadoop2.6.0Eclipsehadoop-eclipse-plugin-2.6.0.jar系统环境及所需文件安装Eclipse安装配置时所用的Eclipse版本是EclipseIDEforJavaEEDevelopers，其下载地址如下：/downloads/download.php?file=/technology/epp/downloads/release/neon/R/eclipse-jee-neon-R-win32-x86_64.zip&mirror_id=448一、配置HadoopMapReduce开发环境安装Eclipse将下载好的Eclipse压缩包解压，即完成安装配置。向Eclipse中添加插件插件包hadoop-eclipse-plugin-2.6.0.jar充当的作用是使Eclipse连接Hadoop集群，方便在Eclipse上面进行hadoopmapreduce方面的开发。该插件包可以在网上搜索关键字进行下载，将下载下来的插件包复制进Eclipse安装目录中的plugins文件夹中，再重启Eclipse即可。在Eclipse中添加map/reduceLocations窗口的方式：依次选择“Window”→“ShowView”→“Other”，打开MapReduceTools选择Map/ReduceLocations然后单击“OK”按钮。这样就能在Eclipse上通过Map/ReduceLocations窗口进行Hadooplocation添加删除和配置等操作。一、配置HadoopMapReduce开发环境通过Eclipse来管理HadoopHDFS修改集群中namenode节点的hdfs-site.xml，添加如下内容:<property><name>dfs.permissions</name><value>false</value></property>在Map/ReduceLocations窗口中添加一个名为hadoop的Hadooplocation然后重启Hadoop集群，即可在Eclipse中管理HDFS，进行一些增删改查看等操作。配置HadoopMapReduce开发环境编写和运行第一个MapReduce程序之前的准备MapReduce应用案例一、编写和运行第一个MapReduce程序之前的准备Windows764位操作系统hadoop2.6.0Eclipsehadoop.dll&winutils.exejavajdk1.8.0系统环境及所需文件建立运行MapReduce程序的依赖（1）下载安装Hadoop（2）下载hadoop.dll&winutils.exe并将其添加到windows下的Hadoop安装目录中的bin目录里。（3）在Eclipse中设置Windows中Hadoop安装目录的绝对路径。“Window”→“Preferences”→“HadoopMap/Reduce”。一、编写和运行第一个MapReduce程序之前的准备在Eclipse中新建一个maven工程；建立编写MapReduce程序的依赖包（1）在EclipseProjectExplorer窗口右击，选择“New”→“Other”；（2）选择“MavenProject”→“Next”；（3）选择相应的maven架包→“Next”；（4）根据实际情况进行配置→“Finish”；在新建的maven工程里，找到pom.xml。在这里可以添加编写MapReduce程序所需要的依赖包。进入/，在搜索栏上查找你所需要添加的依赖包，选择对应的Hadoop版本号，然后单击进入。将其中Maven选项栏里的内容复制粘贴到pom.xml中保存，即可自动下载。一、编写和运行第一个MapReduce程序之前的准备下载完成后的结果建立编写MapReduce程序的依赖包编写和运行第一个MapReduce程序之前的准备MapReduce应用案例配置HadoopMapReduce开发环境三、MapReduce应用案例MapReduce应用案例数据去重单表关联单词计数多表关联排序三、MapReduce应用案例在HDFS根目录下有一文件words，其内容为：helloahell0bhelloc单词计数主要其程序代码如下：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应用案例单词计数主要其程序代码如下：publicstaticvoidmain(String[]args)throwsException{

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

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

"hdfs://master:9000/out" }; }对代码执行RunonHadoop操作，执行结果如右图三、MapReduce应用案例数据去重两个文本文件如下：现通过使用MapReduce去除这两个文件中重复的部分。主要程序代码如下//实现map函数publicvoidmap(Objectkey,Textvalue,Contextcontext)

throwsIOException,InterruptedException{

line=value;

context.write(line,newText(""));}//reduce将输入中的key复制到输出数据的key上，并直接输出

publicstaticclassReduceextendsReducer<Text,Text,Text,Text>{

//实现reduce函数

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

throwsIOException,InterruptedException{

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

}

}

三、MapReduce应用案例数据去重对代码执行RunonHadoop操作，执行结果如下图三、MapReduce应用案例排序两个文本文件如下：现通过使用MapReduce去除这两个文件中重复的部分。主要程序代码如下//实现map函数

publicvoidmap(Objectkey,Textvalue,Contextcontext)

throwsIOException,InterruptedException{

Stringline=value.toString();

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

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

}//reduce将输入中的key复制到输出数据的key上，

//然后根据输入的value-list中元素的个数决定key的输出次数

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

publicstaticclassReduceextendsReducer<IntWritable,IntWritable,IntWritable,IntWritable>{privatestaticIntWritablelinenum=newIntWritable(1);三、MapReduce应用案例排序//实现reduce函数

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();

//这句话很关键

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

job.setJarByClass(DataSort.class);对代码执行RunonHadoop操作，执行结果如下图三、MapReduce应用案例单表关联在HDFS根目录下family_in文件目录中创建包含5-1所示的child_parent（孩子父母）表中内容的文件，要求通过MapReduce输出如表5-2所示的grandchild_grandparent（孙子爷奶）表中的内容。三、MapReduce应用案例单表关联主要其程序代码如下：publicstaticclassMapextendsMapper<Object,Text,Text,Text>{//实现map函数publicvoidmap(Objectkey,Textvalue,Contextcontext)throwsIOException,InterruptedException{Stringchildname=newString();//孩子名称Stringparentname=newString();//父母名称Stringrelationtype=newString();//左右表标识

//输入的一行预处理文本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应用案例单表关联主要其程序代码如下：

publicstaticclassReduceextendsReducer<Text,Text,Text,Text>{//实现reduce函数

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应用案例单表关联intlen=record.length();inti=2;if(0==len){

continue;}

//取得左右表标识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++;}