MapReduce 基础实战

---

第1关：成绩统计

编程要求
使用MapReduce计算班级每个学生的最好成绩，输入文件路径为/user/test/input，请将计算后的结果输出到/user/test/output/目录下。

测试说明
输入文件在你每次点击评测的时候，平台会为你创建，无需你自己创建，只需要启动HDFS，编写java代码即可。

输入文件的数据格式如下：
张三 12
李四 13
张三 89
李四 92
…

依照如上格式你应该输出：

张三 89
李四 92

具体本关的预期输出请查看右侧测试集。

因为大数据实训消耗资源较大，且map/reduce运行比较耗时，所以评测时间较长，大概在60秒左右，请耐心等待。

开始你的任务吧，祝你成功！

代码示例入下：

import java.io.IOException;
import java.util.StringTokenizer;
 
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
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 WordCount {
    /********** Begin **********/
    public static class TokenizerMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();
        private int maxValue = 0;
        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            StringTokenizer itr = new StringTokenizer(value.toString(),"\n");
            while (itr.hasMoreTokens()) {
                String[] str = itr.nextToken().split(" ");
                String name = str[0];
                one.set(Integer.parseInt(str[1]));
                word.set(name);
                context.write(word,one);
            }
            //context.write(word,one);
        }
    }
    public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();
        public void reduce(Text key, Iterable<IntWritable> values, Context context)
                throws IOException, InterruptedException {
            int maxAge = 0;
            int age = 0;
            for (IntWritable intWritable : values) {
                maxAge = Math.max(maxAge, intWritable.get());
            }
            result.set(maxAge);
            context.write(key, result);
        }
    }
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = new Job(conf, "word count");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        String inputfile = "/user/test/input";
        String outputFile = "/user/test/output/";
        FileInputFormat.addInputPath(job, new Path(inputfile));
        FileOutputFormat.setOutputPath(job, new Path(outputFile));
        job.waitForCompletion(true);
    /********** End **********/
    }
}

还需在命令行启动：

start-all.sh

第2关：文件内容合并去重

编程要求
接下来我们通过一个练习来巩固学习到的MapReduce知识吧。

对于两个输入文件，即文件file1和文件file2，请编写MapReduce程序，对两个文件进行合并，并剔除其中重复的内容，得到一个新的输出文件file3。
为了完成文件合并去重的任务，你编写的程序要能将含有重复内容的不同文件合并到一个没有重复的整合文件，规则如下：

第一列按学号排列；
学号相同，按x,y,z排列；
输入文件路径为：/user/tmp/input/；
输出路径为：/user/tmp/output/。
注意：输入文件后台已经帮你创建好了，不需要你再重复创建。

测试说明
程序会对你编写的代码进行测试：
输入已经指定了测试文本数据：需要你的程序输出合并去重后的结果。
下面是输入文件和输出文件的一个样例供参考。

输入文件file1的样例如下：
20150101 x
20150102 y
20150103 x
20150104 y
20150105 z
20150106 x

输入文件file2的样例如下：
20150101 y
20150102 y
20150103 x
20150104 z
20150105 y

根据输入文件file1和file2合并得到的输出文件file3的样例如下：

20150101 x
20150101 y
20150102 y
20150103 x
20150104 y
20150104 z
20150105 y
20150105 z
20150106 x

开始你的任务吧，祝你成功！

示例代码如下：

import java.io.IOException;
import java.util.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
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上 注意在map方法中要抛出异常：throws IOException,InterruptedException
    /********** Begin **********/
    public static class Map extends Mapper<LongWritable, Text, Text, Text >
    {
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            String str = value.toString();
            String[] data = str.split(" ");
            Text t1= new Text(data[0]);
            Text t2 = new Text(data[1]);
            context.write(t1,t2);
        }
    } 
    /********** End **********/
    
    //在这重载reduce函数，直接将输入中的key复制到输出数据的key上  注意在reduce方法上要抛出异常：throws IOException,InterruptedException
    /********** Begin **********/
    public static class Reduce  extends Reducer<Text, Text, Text, Text>
    {
        protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            List<String> list = new ArrayList<>();
            for (Text text : values) {
                String str = text.toString();
                if(!list.contains(str)){
                    list.add(str);
                }
            }
            Collections.sort(list);
            for (String text : list) {
                context.write(key, new Text(text));
            }
        }
    /********** End **********/
    }
    
    public static void main(String[] args) throws Exception{
        Configuration conf = new Configuration();
         Job job = new Job(conf, "word count");
        job.setJarByClass(Merge.class);
        job.setMapperClass(Map.class);
        job.setCombinerClass(Reduce.class);
        job.setReducerClass(Reduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        String inputPath = "/user/tmp/input/";  //在这里设置输入路径
        String outputPath = "/user/tmp/output/";  //在这里设置输出路径
        FileInputFormat.addInputPath(job, new Path(inputPath));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

在命令行输入：

start-all.sh

---