大数据 MapReduce编程实践（1）编程实现文件合并和去重操作 |
您所在的位置：网站首页 › 惠普m1130扫描多页怎么合并为1个文件 › 大数据 MapReduce编程实践（1）编程实现文件合并和去重操作 |
大数据 MapReduce编程实践（1）编程实现文件合并和去重操作
|
一、实验目的
1.通过实验掌握基本的MapReduce编程方法； 2.掌握用MapReduce解决一些常见的数据处理问题，包括数据去重、数据排序和数据挖掘等。 二、实验平台已经配置完成的Hadoop伪分布式环境。 三. 实验内容和要求 （1）编程实现文件合并和去重操作对于两个输入文件，即文件A和文件B，请编写MapReduce程序，对两个文件进行合并，并剔除其中重复的内容，得到一个新的输出文件C。下面是输入文件和输出文件的一个样例供参考。 输入文件A的样例如下： 20150101 x 20150102 y 20150103 x 20150104 y 20150105 z 20150106 x 输入文件B的样例如下： 20150101 y 20150102 y 20150103 x 20150104 z 20150105 y 根据输入文件A和B合并得到的输出文件C的样例如下： 20150101 x 20150101 y 20150102 y 20150103 x 20150104 y 20150104 z 20150105 y 20150105 z 20150106 x 需要首先删除HDFS中与当前Linux用户hadoop对应的input和output目录（即HDFS中的“/user/hadoop/input”和“/user/hadoop/output”目录），这样确保后面程序运行不会出现问题 cd /usr/local/hadoop ./bin/hdfs dfs -rm -r input ./bin/hdfs dfs -rm -r output然后，再在HDFS中新建与当前Linux用户hadoop对应的input目录，即“/user/hadoop/input”目录 cd /usr/local/hadoop ./bin/hdfs dfs -mkdir input创建A.txt B.txt,输入上述内容 vi A.txt vi B.txt将A,B上传到HDFS中 cd /usr/local/hadoop ./bin/hdfs dfs -put ./A.txt input ./bin/hdfs dfs -put ./B.txt input运行代码： import java.io.IOException; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser; public class Merge { /** * @param args * 对A,B两个文件进行合并，并剔除其中重复的内容，得到一个新的输出文件C */ //重载map函数，直接将输入中的value复制到输出数据的key上 public static class Map extends Mapper{ private static Text text = new Text(); public void map(Object key, Text value, Context context) throws IOException,InterruptedException{ text = value; context.write(text, new Text("")); } } //重载reduce函数，直接将输入中的key复制到输出数据的key上 public static class Reduce extends Reducer{ public void reduce(Text key, Iterable values, Context context ) throws IOException,InterruptedException{ context.write(key, new Text("")); } } public static void main(String[] args) throws Exception{ // TODO Auto-generated method stub Configuration conf = new Configuration(); conf.set("fs.default.name","hdfs://localhost:9000"); String[] otherArgs = new String[]{"input","output"}; /* 直接设置输入参数 */ if (otherArgs.length != 2) { System.err.println("Usage: wordcount "); System.exit(2); } Job job = Job.getInstance(conf,"Merge and duplicate removal"); job.setJarByClass(Merge.class); job.setMapperClass(Map.class); job.setCombinerClass(Reduce.class); job.setReducerClass(Reduce.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0])); FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }编译运行：
现在有多个输入文件，每个文件中的每行内容均为一个整数。要求读取所有文件中的整数，进行升序排序后，输出到一个新的文件中，输出的数据格式为每行两个整数，第一个数字为第二个整数的排序位次，第二个整数为原待排列的整数。下面是输入文件和输出文件的一个样例供参考。 输入文件1的样例如下： 33 37 12 40 输入文件2的样例如下： 4 16 39 5 输入文件3的样例如下： 1 45 25 根据输入文件1、2和3得到的输出文件如下： 1 1 2 4 3 5 4 12 5 16 6 25 7 33 8 37 9 39 10 40 11 45 需要首先删除HDFS中与当前Linux用户hadoop对应的input和output目录（即HDFS中的“/user/hadoop/input”和“/user/hadoop/output”目录），这样确保后面程序运行不会出现问题 cd /usr/local/hadoop ./bin/hdfs dfs -rm -r input ./bin/hdfs dfs -rm -r output然后，再在HDFS中新建与当前Linux用户hadoop对应的input目录，即“/user/hadoop/input”目录 cd /usr/local/hadoop ./bin/hdfs dfs -mkdir input创建test1.txt test2.txt test3,输入上述内容 vi test1.txt vi test2.txt vi test3.txt将test1,test2,test3上传到HDFS中 cd /usr/local/hadoop ./bin/hdfs dfs -put ./test1.txt input ./bin/hdfs dfs -put ./test2.txt input ./bin/hdfs dfs -put ./test3.txt input运行代码： import java.io.IOException; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Partitioner; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser; public class Merge { /** * @param args * 输入多个文件，每个文件中的每行内容均为一个整数 * 输出到一个新的文件中，输出的数据格式为每行两个整数，第一个数字为第二个整数的排序位次，第二个整数为原待排列的整数 */ //map函数读取输入中的value，将其转化成IntWritable类型，最后作为输出key public static class Map extends Mapper{ private static IntWritable data = new IntWritable(); public void map(Object key, Text value, Context context) throws IOException,InterruptedException{ String text = value.toString(); data.set(Integer.parseInt(text)); context.write(data, new IntWritable(1)); } } //reduce函数将map输入的key复制到输出的value上，然后根据输入的value-list中元素的个数决定key的输出次数,定义一个全局变量line_num来代表key的位次 public static class Reduce extends Reducer{ private static IntWritable line_num = new IntWritable(1); public void reduce(IntWritable key, Iterable values, Context context) throws IOException,InterruptedException{ for(IntWritable val : values){ context.write(line_num, key); line_num = new IntWritable(line_num.get() + 1); } } } //自定义Partition函数，此函数根据输入数据的最大值和MapReduce框架中Partition的数量获取将输入数据按照大小分块的边界，然后根据输入数值和边界的关系返回对应的Partiton ID public static class Partition extends Partitioner{ public int getPartition(IntWritable key, IntWritable value, int num_Partition){ int Maxnumber = 65223;//int型的最大数值 int bound = Maxnumber/num_Partition+1; int keynumber = key.get(); for (int i = 0; i |
查看实验结果：
【本文地址】
今日新闻 |
推荐新闻 |