上一篇文章我们了解了MapReduce优化方面的知识,现在我们通过简单的项目,学会如何优化MapReduce性能

1、项目介绍

  我们使用简单的成绩数据集,统计出0~20、20~50、50~100这三个年龄段的男、女学生的最高分数

2、数据集

  姓名     年龄  性别   成绩

  Alice     23   female  45

  Bob      34   male   89

  Chris    67   male   97

  Kristine   38   female  53

  Connor    25   male   27

  Daniel    78   male   95

  James    34   male   79

  Alex      52  male   69

3、分析

  基于需求,我们通过以下几步完成:

  1、编写Mapper类,按需求将数据集解析为key=gender,value=name+age+score,然后输出

  2、编写Partitioner类,按年龄段,将结果指定给不同的Reduce执行

  3、编写Reducer类,分别统计出男女学生的最高分

  4、编写run方法执行MapReduce作业

4、实现

package com.buaa; 
import java.io.IOException; 
 
import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.conf.Configured; 
import org.apache.hadoop.fs.FileSystem; 
import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.io.LongWritable; 
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.Tool; 
import org.apache.hadoop.util.ToolRunner; 
 
/**  * @ProjectName BestScoreCount 
* @PackageName com.buaa 
* @ClassName Gender 
* @Description 统计不同年龄段内,男、女最高分数 
* @Author 刘吉超 
* @Date 2016-05-09 09:49:50 
*/ public class Gender extends Configured implements Tool { private static String TAB_SEPARATOR = "\t";  
     public static class GenderMapper extends Mapper<LongWritable, Text, Text, Text> { /* 
         * 调用map解析一行数据,该行的数据存储在value参数中,然后根据\t分隔符,解析出姓名,年龄,性别和成绩 
         */ public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { /* 
             * 姓名            年龄        性别        成绩 
             * Alice           23         female     45 
             * 每个字段的分隔符是tab键 
             */ // 使用\t,分割数据 
            String[] tokens = value.toString().split(TAB_SEPARATOR); 
             // 性别 
            String gender = tokens[2]; // 姓名    年龄    成绩 
            String nameAgeScore = tokens[0] + TAB_SEPARATOR + tokens[1] + TAB_SEPARATOR + tokens[3]; 
             // 输出  key=gender  value=name+age+score 
            context.write(new Text(gender), new Text(nameAgeScore)); 
        } 
    } 
     /* 
     * 合并 Mapper输出结果 
     */ public static class GenderCombiner extends Reducer<Text, Text, Text, Text>  { 
 public void reduce(Text key, Iterable<Text> values, Context context)throws IOException, InterruptedException { int maxScore = Integer.MIN_VALUE; int score = 0; 
            String name = " "; 
            String age = " "; 
             for (Text val : values) { 
                String[] valTokens = val.toString().split(TAB_SEPARATOR); 
                score = Integer.parseInt(valTokens[2]); if (score > maxScore) { 
                    name = valTokens[0]; 
                    age = valTokens[1]; 
                    maxScore = score; 
                } 
            } 
             
            context.write(key, new Text(name + TAB_SEPARATOR + age + TAB_SEPARATOR + maxScore)); 
        } 
    } 
     /* 
     * 根据 age年龄段将map输出结果均匀分布在reduce上 
     */ public static class GenderPartitioner extends Partitioner<Text, Text> { 
 
        @Override public int getPartition(Text key, Text value, int numReduceTasks) { 
            String[] nameAgeScore = value.toString().split(TAB_SEPARATOR); // 学生年龄 int age = Integer.parseInt(nameAgeScore[1]); 
 // 默认指定分区 0 if (numReduceTasks == 0) return 0; 
 // 年龄小于等于20,指定分区0 if (age <= 20) { return 0; 
            }else if (age <= 50) { // 年龄大于20,小于等于50,指定分区1 return 1 % numReduceTasks; 
            }else // 剩余年龄,指定分区2 return 2 % numReduceTasks; 
        } 
    } 
 /* 
     * 统计出不同性别的最高分 
     */ public static class GenderReducer extends Reducer<Text, Text, Text, Text>  { 
        @Override public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { int maxScore = Integer.MIN_VALUE; int score = 0; 
            String name = " "; 
            String age = " "; 
            String gender = " "; 
             // 根据key,迭代 values集合,求出最高分 for (Text val : values) { 
                String[] valTokens = val.toString().split(TAB_SEPARATOR); 
                score = Integer.parseInt(valTokens[2]); if (score > maxScore) { 
                    name = valTokens[0]; 
                    age = valTokens[1]; 
                    gender = key.toString(); 
                    maxScore = score; 
                } 
            } 
             
            context.write(new Text(name), new Text("age:" + age + TAB_SEPARATOR + "gender:" + gender + TAB_SEPARATOR + "score:" + maxScore)); 
        } 
    } 
 
    @SuppressWarnings("deprecation") 
    @Override public int run(String[] args) throws Exception { // 读取配置文件 
        Configuration conf = new Configuration(); 
 
        Path mypath = new Path(args[1]); 
        FileSystem hdfs = mypath.getFileSystem(conf); if (hdfs.isDirectory(mypath)) { 
            hdfs.delete(mypath, true); 
        } 
         // 新建一个任务 
        Job job = new Job(conf, "gender"); // 主类 
        job.setJarByClass(Gender.class); // Mapper 
        job.setMapperClass(GenderMapper.class); // Reducer 
        job.setReducerClass(GenderReducer.class); 
 // map 输出key类型 
        job.setMapOutputKeyClass(Text.class); // map 输出value类型 
        job.setMapOutputValueClass(Text.class); 
         // reduce 输出key类型 
        job.setOutputKeyClass(Text.class); // reduce 输出value类型 
        job.setOutputValueClass(Text.class); 
 // 设置Combiner类 
        job.setCombinerClass(GenderCombiner.class); 
 // 设置Partitioner类 
        job.setPartitionerClass(GenderPartitioner.class); // reduce个数设置为3 
        job.setNumReduceTasks(3); 
 // 输入路径 
        FileInputFormat.addInputPath(job, new Path(args[0])); // 输出路径 
        FileOutputFormat.setOutputPath(job, new Path(args[1])); 
         // 提交任务 return job.waitForCompletion(true)?0:1; 
    } 
     public static void main(String[] args) throws Exception { 
        String[] args0 = { "hdfs://ljc:9000/buaa/gender/gender.txt", "hdfs://ljc:9000/buaa/gender/out/" 
        }; int ec = ToolRunner.run(new Configuration(),new Gender(), args0); 
        System.exit(ec); 
    } 
}

5、运行效果

  image

发布评论

分享到:

IT虾米网

微信公众号号:IT虾米 (左侧二维码扫一扫)欢迎添加!

MapReduce明星搜索指数统计,找出人气王详解
你是第一个吃螃蟹的人
发表评论

◎欢迎参与讨论,请在这里发表您的看法、交流您的观点。