Hadoop 实战 | 词频统计WordCount

词频统计

通过分析大量文本数据中的词频，可以识别常见词汇和短语，从而抽取文本的关键信息和概要，有助于识别文本中频繁出现的关键词，这对于理解文本内容和主题非常关键。同时，通过分析词在文本中的相对频率，可以帮助理解词在不同上下文中的含义和语境。

"纽约时报"评论数据集记录了有关《纽约时报》2017年1月至5月和2018年1月至4月发表的文章上的评论的信息。月度数据分为两个csv文件：一个用于包含发表评论的文章，另一个用于评论本身。评论的csv文件总共包含超过200万条评论，有34个特征，而文章的csv文件包含超过9000篇文章，有16个特征。

本实验需要提取其中的 articleID 和 snippet 字段进行词频统计

MapReduce

在Hadoop中，输入文件通常会通过InputFormat被分成一系列的逻辑分片，分片是输入文件的逻辑划分，每个分片由一个Mapper处理。

本实验中，WordCount通过MapReduce统计snippet 字段中每个单词出现的总次数。程序主要包括Mapper, Reducer, Driver三个部分。

自定义的Mapper和Reducer都要继承各自的父类。Mapper中的业务逻辑写在map()方法中，Reducer的业务逻辑写在reduce()方法中。整个程序还需要一个Driver来进行提交，提交的是一个描述了各种必要信息的job对象。

程序总体流程如下图所示。

Mapper

Mapper的主要任务是处理输入分片并生成中间键值对，这些键值对将被传递给Reducer进行进一步处理，也就是对应的Map的过程。

在本实验中，Mapper需要将这行文本中的单词提取出来，针对每个单词输出一个<word, 1>的<Key, Value>对。之后MapReduce会对这些<word,1>进行排序重组，将相同的word放在一起，形成<word, [1,1,1,1,1,1,1…]>的<Key,Value >结构并传递给Reducer。

Reducer

Reducer则以中间键值对为输入，将其按照键进行分组，并将每个组的值按一定规则合并成最终的输出。

注意在此阶段前，Hadoop框架会自行将中间键值对经过默认的排序分区分组，Key相同的单词会作为一组数据构成新的<Key, Value>对。

在本实验中，Reducer将集合里的1求和，再将单词（word）与这个和（sum）组成一个<Key, Value>，也就是<word, sum>输出。每一个输出就是一个单词和它的词频统计总和了。

Driver

Driver是一个程序的主入口，负责配置和启动整个MapReduce任务。Driver类通常包含了整个MapReduce作业的配置信息、作业的输入路径、输出路径等信息，并启动MapReduce作业的执行。

总结

该程序基于Hadoop MapReduce框架实现了简单的单词计数功能，适用于大规模文本数据的并行处理。

PSEUDO-CODE 2?? WordCount(词频统计)

/* Map函数，处理每一行的文本 */

1：input <Key,Value>；?????????????????????????????????? //Value使用Text类型表示文本行

2：从文本中提取文档ID和实际文本内容snippet；

3：使用空格、单引号和破折号作为分隔符，将文本snippet分词；

4：for 文本snippet中的每个单词：

5：???? 去除特殊字符后将<word,1>写入context，发射给Reducer；

6：end for

/* Reduce函数，处理相同键的所有值 */

1：input <Key,Value>，sum←0；???????????? //来自Map的<word,[1,1,1…]>

2：for Value的每个1：

3：???? 累加计数sum += 1；

代码

import java.io.IOException;
import java.util.regex.*;
import java.util.Iterator;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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 {
    public WordCount() {
    }
     public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = (new GenericOptionsParser(conf, args)).getRemainingArgs();
        if(otherArgs.length < 2) {
            System.err.println("Usage: wordcount <in> [<in>...] <out>");
            System.exit(2);
        }
        Job job = Job.getInstance(conf, "word count");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(WordCount.TokenizerMapper.class);
        job.setCombinerClass(WordCount.IntSumReducer.class);
        job.setReducerClass(WordCount.IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class); 
        for(int i = 0; i < otherArgs.length - 1; ++i) {
            FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
        }
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1]));
        System.exit(job.waitForCompletion(true)?0:1);
    }

     public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
    	    private static final IntWritable one = new IntWritable(1);
    	    private Text word = new Text();

    	    public TokenizerMapper() {
    	    }

    	    public void map(Object key, Text value, Mapper<Object, Text, Text, IntWritable>.Context context)
    	            throws IOException, InterruptedException {
    	        // Split DocID and the actual text
    	        String DocId = value.toString().substring(0, value.toString().indexOf("\t"));
    	        String value_raw =  value.toString().substring(value.toString().indexOf("\t") + 1);
    	        
    	        // Reading input one line at a time and tokenizing by using space, "'", and "-" characters as tokenizers.
    	        StringTokenizer itr = new StringTokenizer(value_raw, " '-");
    	        
    	        // Iterating through all the words available in that line and forming the key/value pair.
    	        while (itr.hasMoreTokens()) {
    	          // Remove special characters
    	          word.set(itr.nextToken().replaceAll("[^a-zA-Z]", ""));
    	          if(word.toString() != "" && !word.toString().isEmpty()){
    	        	  context.write(word, one);
    	          }
    	    }
    	}
     }
	public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
	        private IntWritable result = new IntWritable();
	        public IntSumReducer() {
	        }
	        public void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
	            int sum = 0;
	            IntWritable val;
	            for(Iterator i$ = values.iterator(); i$.hasNext(); sum += val.get()) {
	                val = (IntWritable)i$.next();
	            }
	            this.result.set(sum);
	            context.write(key, this.result);
	        }
	    }
	}