Hadoop基础（五十六）：其他面试题 手写Hadoop WordCount

环境说明:

• jdk1.8
• hadoop-2.7.7
• windows上配置好的带有hadoop环境的eclipse

1.自定义Mapper

/*
 * LongWritable对应输入的key类型，默认是行的偏移量LongWritable
 * Text,对应上输入的value类型，默认行数据Text
 * Text:对应输出的key类型，不能使用默认值，需要根据需求更改
 * Text:对应输出的value类型，根据需求修改
 * @author lesie
 * 要求输出的格式(key,1)
 * 单词计数输出的key类型为Text
 * 输出的value类型为IntWritable
 */
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
    
    /*
     * KEYIN
     * VALUEIN
     * context--环境对象，输出结果
     * @see org.apach.hadoop.mapreduce.Mapper#map(KEYIN,VALUEIN,...)
     */
    public void map(LongWritable ikey,Text ivalue,Context context) throws IOException, InterruptedException
    {
        //获取一行数据
        String line=ivalue.toString();
        
        //按空格切片
        String []arrs=line.split(" ");
        for(String arr:arrs)
        {
            context.write(new Text(arr),new IntWritable(1));
        }
        
    }
}

2.自定义Reducer

/*
 * reducer的数输入key用公式mapper输出的key类型
 * valuein:reducer的输入value应该是mapper输出的value类型
 * keyout:根据业务而定
 * valueout:根据业务而定
 * @author lesie 
 * 工作机制:
 * 1.将key相同的value进行合并，形成一个Iterable，交给程序
 * eg:(hello,<1,1,1,1,1,1>)
 * 2.reduce方法执行的次数取决于mapper输出的key，有多个不同的key执行多少次
 * 3.默认的排序，对key进行排序，先按照数字进行排再按照字典顺序
 */

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    
    public void reduce(Text _key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        // process values
        //定义计数变量
        int sum=0;
        //进行累加操作
        for (IntWritable val : values) {
            //通过get方法取出其中的值
            sum+=val.get();
        }
        //输出数据,最终结果,key是单词Text,value是单词出现的总次数
        context.write(_key, new IntWritable(sum));
    }

}

3.主程序

public class WordCountDriver {

    public static void main(String[] args) throws Exception {
        // TODO Auto-generated method stub
        //获取当前配置
        Configuration conf=new Configuration();
        
        //获取一个表示当前Mapreduce作业的Job对象，向ahdoop申请一个job任务执行逻辑
        Job job=Job.getInstance();
        
        //指定程序入口
        job.setJarByClass(WordCountDriver.class);
        
        //设置需要执行的Mapper类
        job.setMapperClass(WordCountMapper.class);
        
        //设置Reducer类
        job.setReducerClass(WordCountReducer.class);
        
        //设置Mapper的输出类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        
        //设置Reducer的输出结果类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        
        //设置输入路径
        FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.140.128:9000/wc/words.txt"));
        
        //设置输出路径
        FileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.140.128:9000/wc/result6"));
        
        //任务的提交
        job.waitForCompletion(true);
    }

}