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 textString 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 charactersword.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);}}}