Hadoop源码分析之WordCount

WordCount程序是hadoop自带的案例，我们可以在 hadoop 解压目录下找到包含这个程序的 jar 文件(hadoop-mapreduce-examples-2.7.1.jar)，该文件所在路径为 hadoop/share/hadoop/mapreduce。
最简单的MapReduce应用程序至少包含 3 个部分：一个 Map 函数、一个 Reduce 函数和一个 main 函数。在运行一个mapreduce计算任务时候，任务过程被分为两个阶段：map阶段和reduce阶段，每个阶段都是用键值对（key/value）作为输入（input）和输出（output）。main 函数将作业控制和文件输入/输出结合起来。
Map过程：并行读取文本，对读取的单词进行map操作，每个词都以<key,value>形式生成。
举例：
一个有三行文本的文件进行MapReduce操作。
1、读取第一行Hello World Bye World ，分割单词形成Map：
<Hello,1> <World,1> <Bye,1> <World,1>
2、读取第二行Hello Hadoop Bye Hadoop ，分割单词形成Map：
<Hello,1> <Hadoop,1> <Bye,1> <Hadoop,1>
3、读取第三行Bye Hadoop Hello Hadoop，分割单词形成Map：
<Bye,1> <Hadoop,1> <Hello,1> <Hadoop,1>
Reduce过程：是对map的结果进行排序，合并，最后得出词频。
举例：
1、经过进一步处理(combiner),将形成的Map根据相同的key组合成value数组：
<Bye,1,1,1> <Hadoop,1,1,1,1> <Hello,1,1,1> <World,1,1>
2、循环执行Reduce(K,V[])，分别统计每个单词出现的次数：
<Bye,3> <Hadoop,4> <Hello,3> <World,2>
WordCount源码如下：

package org.apache.hadoop.examples;

import java.io.IOException;
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 {
/**
* TokenizerMapper继承Mapper类
* Mapper<KEYIN（输入key类型）, VALUEIN（输入value类型）, KEYOUT（输出key类型）, VALUEOUT（输出value类型）>
*/
public static class TokenizerMapper
extends Mapper<Object, Text, Text, IntWritable>{

// 因为若每个单词出现后，就置为 1，并将其作为一个<key,value>对，因此可以声明为常量，值为 1
private final static IntWritable one = new IntWritable(1);//VALUEOUT
private Text word = new Text();//KEYOUT

/**
*重写map方法，读取初试划分的每一个键值对，即行偏移量和一行字符串，key为偏移量，value为该行字符串
*/
public void map(Object key, Text value, Context context
) throws IOException, InterruptedException {
/**
* 因为每一行就是一个spilt，并会为之生成一个mapper，所以我们的参数，key就是偏移量，value就是一行字符串
* value是一行的字符串，这里将其切割成多个单词,将每行的单词进行分割,按照"  \t\n\r\f"(空格、制表符、换行符、回车符、换页)进行分割
*/
StringTokenizer itr = new StringTokenizer(value.toString());

//遍历
while (itr.hasMoreTokens()) {
//获取每个值并设置map输出的key值
word.set(itr.nextToken());
//
1d25c
one代表1，最开始每个单词都是1次，context直接将<word,1>写到本地磁盘上
//write函数直接将两个参数封装成<key,value>
context.write(word, one);
}
}
}

/**
* IntSumReducer继承Reducer类
* Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>:Map的输出类型，就是Reduce的输入类型
*/
public static class IntSumReducer
extends Reducer<Text,IntWritable,Text,IntWritable> {
//输出结果，总次数
private IntWritable result = new IntWritable();

/**
*重写reduce函数，key为单词，values是reducer从多个mapper中得到数据后进行排序并将相同key组
*合成<key.list<V>>中的list<V>，也就是说明排序这些工作都是mapper和reducer自己去做的，
*我们只需要专注与在map和reduce函数中处理排序处理后的结果
*/
public void reduce(Text key, Iterable<IntWritable> values,
Context context
) throws IOException, InterruptedException {

/**
*因为在同一个spilt对应的mapper中，会将其进行combine，使得其中单词（key）不重复，然后将这些键值对按照
*hash函数分配给对应的reducer，reducer进行排序，和组合成list，然后再调用的用户自定义的函数
*/
int sum = 0;//累加器，累加每个单词出现的次数
//遍历values
for (IntWritable val : values) {
sum += val.get();//累加
}
result.set(sum);//设置输出value
context.write(key, result);//context输出reduce结果
}
}

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，设置名称
job.setJarByClass(WordCount.class);//1、设置job运行的类
//2、设置mapper类、Combiner类和Reducer类
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
//3、设置输出结果key和value的类
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
for (int i = 0; i < otherArgs.length - 1; ++i) {
FileInputFormat.addInputPath(job, new Path(otherArgs[i]));//4、为job设置输入陆军
}
FileOutputFormat.setOutputPath(job,//5、为job设置输出路径
new Path(otherArgs[otherArgs.length - 1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);//6、结束程序
}
}

转自链接：https://www.jianshu.com/p/3136a9fa84ed