MapReduce的基础案例（一）WordCount，词频统计

迈不过友情╰ 2024-04-18 11:41 78阅读 0赞

文本文档words.txt

hello tom
    hello lina
    hello tom
    hello GPY
    HI selina

结果样式：

GPY	1
    HI	1
    hello	4
    lina	1
    selina	1
    tom	2

Java代码：

package MR;
    
    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.LongWritable;
    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;
    
    public class WC{
        
    //定义Mapper类，四个参数分别是（map读入的偏移量类型，值类型，map即将输出的键类型值类型）
        public static  class MyMap extends Mapper<LongWritable, Text, Text, IntWritable>{
        
            @Override
            protected void map(LongWritable key, Text value,Context context)
                    throws IOException, InterruptedException {
        
                String[] fields = value.toString().split(" ");//把读入的一行数据，按空格切割
                for(String s: fields){
        
                    context.write(new Text(s), new IntWritable(1));//每一单词都作为键值，并赋予value1，作为词频的基本数
                }
            }
        }
    
    //定义reducer类，四个参数分别是（map输出的键类型，值类型，reducer即将输出的键类型值类型）
        public static class MyReduce extends Reducer<Text, IntWritable, Text, IntWritable>{
        
            @Override
            protected void reduce(Text key, Iterable<IntWritable> values,//此时经过shuffle已经将value变为数组
                                  Context context) throws IOException, InterruptedException {
        
                int count=0;
                for(IntWritable i:values){
        
                    count+=i.get();//在ruducer里把每个值相加
                }
                context.write(key, new IntWritable(count));//
            }
        }
        public static void main(String[] args) {
        
            Configuration conf=new Configuration();
            try {
        
                Job job=Job.getInstance(conf);
                job.setJarByClass(WC.class);//设置主类
                job.setMapperClass(MyMap.class);//加载map类
                job.setReducerClass(MyReduce.class);//加载reduce类
    
                job.setOutputKeyClass(Text.class);//输出的KEY格式
                job.setOutputValueClass(IntWritable.class);//输出的value格式
    
                Path inPath =new Path("/input/words.txt");
                FileInputFormat.addInputPath(job, inPath);
    
                Path outpath=new Path("/output/WordCount/result");
                if(outpath.getFileSystem(conf).exists(outpath)){
        
                    outpath.getFileSystem(conf).delete(outpath, true);
                }
                FileOutputFormat.setOutputPath(job, outpath);
    
                job.waitForCompletion(true);
            } catch (Exception e) {
        
                // TODO Auto-generate
                //  .d catch block
                e.printStackTrace();
            }
        }
    }