Mapreduce经典案例：词频统计

/**
 * Mapper类
 * 示例: hello hadoop hadoop --> (hello, 1), (hadoop, 1), (hadoop, 1)
 */
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private static final Text KEY = new Text();
    private static final IntWritable VAL = new IntWritable(1); // 指定value为数字1
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 将数据行转为字符串
        String line = value.toString();
        // 指定分隔符并分割数据
        String[] words = line.split(" ");
        // 将数据写入下一阶段
        for (String word : words) {
            KEY.set(word);
            context.write(KEY, VAL);
        }
    }
}

2、WordCountReducer.java

/**
 * Reducer类
 * 示例: (hello, 1), (hadoop, 1), (hadoop, 1) --> (hello, 1), (hadoop, 2)
 */
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    private static final IntWritable VAL = new IntWritable();
    protected void reduce(Text KEY, Iterable<IntWritable> value, Context context) throws IOException, InterruptedException {
        // 定义计数器
        int counter = 0;
        // 将相同的单词累计
        for (IntWritable num : value) {
            counter += num.get();
            // 或者使用 counter ++ ;
        }
        // 设置value值并写入下一阶段
        VAL.set(counter);
        context.write(KEY, VAL);
    }
}

3、WordCountMain.java

public class WordCountMain {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        // 实例化任务
        Job job = Job.getInstance(new Configuration());
        // 设置启动类
        job.setJarByClass(WordCountMain.class);
        // 设置Mapper类以及Map的输出类型
        job.setMapperClass(WordCountMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        // 设置Reducer类以及Reduce的输出类型
        job.setReducerClass(WordCountReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        // 设置文件输入输出路径
        FileInputFormat.setInputPaths(job, args[0]);
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        // 任务执行
        boolean res = job.waitForCompletion(true);
        System.exit(res ? 0 : 1);
    }
}

以上为WordCount经典案例，下面介绍一个词频统计的应用场景：统计网站日志数据每分钟的浏览量。

public class weblogMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private static final Text KEY = new Text();
    private static final IntWritable VAL = new IntWritable(1); // 指定value为数字1
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 将数据行转为字符串
        String line = value.toString();
        // 获取时间字段
        String date = line.split(" ")[3].substring(1);
        // 将时间中的秒数去除
        String date2 = date.substring(0, date.length() - 3);
        // 设置KEY
        KEY.set(date2);
        // 将数据写入下一阶段
        context.write(KEY, VAL);
    }
}

本文介绍了词频统计经典案例，并在其基础上介绍了它的实际应用场景，以上就是本期分享的全部内容，想要获取本文中的代码及全部数据，请在公众号后台回复关键词20230119
，若各位小伙伴有什么不懂的问题也可以直接在公众号后台回复，感谢观看！