实验3-MapReduce编程初级实践

1、 欢迎下载实验3MapReduce编程初级实践实验目的通过实验掌握基本的MapReduce编程方法；掌握用MapReduce解决一些常见的数据处理问题，包括数据去重、数据排序和数据挖掘等。实验平台已经配置完成的Hadoop伪分布式环境。实验内容和要求1.编程实现文件合并和去重操作对于两个输入文件，即文件A和文件B,请编写MapReduce程序，对两个文件进行合并，并剔除其中重复的内容，得到一个新的输出文件Co下面是输入文件和输出文件的一个样例供参考。实验最终结果（合并的文件）：JU叵J审u軍T眷J皿*Merge.java国hdf5:/!caIhst000/u5er/w曽hdf5：/hcalho

2、st:900D/u520150101y20150101X20150102y|20150102y20150103X20150103X20150104220150104y20150105y20150105z20150106X代码如下：packagecom.Merge;importjava.io.IOException;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.J

3、ob;importorg.apache.hadoop.mapreduce.Mapper;importorg.apache.hadoop.mapreduce.Reducer;importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;publicclassMergepublicstaticclassMapextendsMapperprivatestaticTexttext=newText();publicvoidm

4、ap(Objectkey,Textvalue,Contextcontext)throwsIOException,InterruptedExceptiontext=value;context.write(text,newText();publicstaticclassReduceextendsReducerpublicvoidreduce(Textkey,Iterablevalues,Contextcontext)throwsIOException,InterruptedExceptioncontext.write(key,newText();publicstaticvoidmain(Strin

5、gargs)throwsExceptionConfigurationconf=newConfiguration();conf.set(fs.defaultFS,hdfs:/localhost:9000);StringotherArgs=newStringinput,output;if(otherArgs.length!=2)System.err.println(Usage:Mergeandduplicateremoval);System.exit(2);Jobjob=Job.getlnstance(conf,Mergeandduplicateremoval);job.setJarByClass

6、(Merge.class);job.setMapperClass(Map.class);job.setReducerClass(Reduce.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(Text.class);FileInputFormat.addInputPath(job,newPath(otherArgs0);FileOutputFormat.setOutputPath(job,newPath(otherArgs1);System.exit(job.waitForCompletion(true)?0:1)

7、;2.编写程序实现对输入文件的排序现在有多个输入文件，每个文件中的每行内容均为一个整数。要求读取所有文件中的整数，进行升序排序后，输出到一个新的文件中，输出的数据格式为每行两个整数，第一个数字为第二个整数的排序位次，第二个整数为原待排列的整数。下面是输入文件和输出文件的一个样例供参考。实验结果截图：代码如下：packagecom.MergeSort;importjava.io.IOException;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop

8、.io.IntWritable;importorg.apache.hadoop.io.Text;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;publiccla

9、ssMergeSortpublicstaticclassMapextendsMapperprivatestaticIntWritabledata=newIntWritable();publicvoidmap(Objectkey,Textvalue,Contextcontext)throwsIOException,InterruptedExceptionStringline=value.toString();data.set(Integer.parseInt(line);context.write(data,newIntWritable(1);publicstaticclassReduceext

10、endsReducerprivatestaticIntWritablelinenum=newIntWritable(1);publicvoidreduce(IntWritablekey,Iterablevalues,Contextcontext)throwsIOException,InterruptedExceptionfor(IntWritableval:values)context.write(linenum,key);linenum=newIntWritable(linenum.get()+1);publicstaticvoidmain(Stringargs)throwsExceptio

11、nConfigurationconf=newConfiguration();conf.set(fs.defaultFS,hdfs:/localhost:9000);StringotherArgs=newStringinput2,output2;/*直接设置输入参数*/if(otherArgs.length!=2)System.err.println(Usage:mergesort);System.exit(2);Jobjob=Job.getInstance(conf,mergesort);job.setJarByClass(MergeSort.class);job.setMapperClass

12、(Map.class);job.setReducerClass(Reduce.class);job.setOutputKeyClass(IntWritable.class);job.setOutputValueClass(IntWritable.class);FileInputFormat.addInputPath(job,newPath(otherArgs0);FileOutputFormat.setOutputPath(job,newPath(otherArgs1);System.exit(job.waitForCompletion(true)?0:1);3.对给定的表格进行信息挖掘下面给

13、出一个child-parent的表格，要求挖掘其中的父子辈关系，给出祖孙辈关系的表格。实验最后结果截图如下：令1炽Blhdfs:/localhost:9DOSSTjoin.javahdFs:/localhost:900氷先grand_child沪nd_parentMarkJesseMarkAlicePhilipJessePhilipAliceJoneJesseJoneAliceStevenJesseStevenAliceStevenFrankStevenMaryJoneFrankJoneMary代码如下：packagecom.join;importjava.io.IOException;imp

14、ortjava.util.*;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Job;importorg.apache.hadoop.mapreduce.Mapper;importorg.apache.hadoop.mapreduce.Reducer;importorg.apache.hadoop.mapreduce.lib.input.FileInputForm

15、at;importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;publicclassSTjoinpublicstaticinttime=0;publicstaticclassMapextendsMapperpublicvoidmap(Objectkey,Textvalue,Contextcontext)throwsIOException,InterruptedExceptionStringchild_name=newString();Stringparent_name=newString();Stringrelation_ty

16、pe=newString();Stringline=value.toString();inti=0;while(line.charAt(i)!=)i+;Stringvalues=line.substring(0,i),line.substring(i+1);if(pareTo(child)!=0)child_name=values0;parent_name=values1;relation_type=1;context.write(newText(values1),newText(relation_type+child_name+parent_name);relation_type=2;con

17、text.write(newText(values0),newText(relation_type+child_name+parent_name);publicstaticclassReduceextendsReducerpublicvoidreduce(Textkey,Iterablevalues,Contextcontext)throwsIOException,InterruptedExceptionif(time=0)context.write(newText(grand_child),newText(grand_parent);time+;intgrand_child_num=0;St

18、ringgrand_child=newString10;intgrand_parent_num=0;Stringgrand_parent=newString10;Iteratorite=values.iterator();while(ite.hasNext()Stringrecord=ite.next().toString();intlen=record.length();inti=2;if(len=0)continue;charrelation_type=record.charAt(0);Stringchild_name=newString();Stringparent_name=newSt

19、ring();while(record.charAt(i)!=+)child_name=child_name+record.charAt(i);i+;i=i+1;while(ilen)parent_name=parent_name+record.charAt(i);i+;if(relation_type=1)grand_childgrand_child_num=child_name;grand_child_num+;elsegrand_parentgrand_parent_num=parent_name;grand_parent_num+;if(grand_parent_num!=0&gran

20、d_child_num!=0)for(intm=0;mgrand_child_num;m+)for(intn=0;ngrand_parent_num;n+)context.write(newText(grand_childm),newText(grand_parentn);publicstaticvoidmain(Stringargs)throwsExceptionConfigurationconf=newConfiguration();conf.set(fs.defaultFS,hdfs:/localhost:9000);StringotherArgs=newStringinput3,output3;if(otherArgs.length!=2)System.err.println(Usage:SingleTableJoin);System.exit(2);Jobjob=Job.getInstance(conf,Singletablejoin);job.se