一、前期准备
1. 运行环境
想要运行WordCount程序，其实可以不需要安装任何的Hadoop软件环境，因为实际上执行计算任务的是Hadoop框架集成的各种jar包。Hadoop启动后的各项进程主要用于支持HDFS的使用，各个节点间的通讯，任务调度等等。所以如果我们只是想测试程序的可用性的话可以只新建一个Java项目，然后集成Hadoop相关的jar包，直接运行程序即可。
这种方式只限于代码测试，因为可以随时修改代码并且执行，结果也可以很方便查看。本文主要讲解MapReduce的运行流程，因此不需要搭建任何Hadoop环境，关于Hadoop任务的提交方式将在其它文章中详细说明。
2. 项目新建
- 首先在IDEA中新建一个Maven项目：
- 修改pom.xml，添加Hadoop相关的依赖：
<dependencies> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>3.3.5</version> </dependency> </dependencies>
二、从WordCount开始
对于Hadoop来说，它的Hello World经典案例当属WordCount了，给出一段文本，我们统计出其中一共包含多少单词。我们可以使用MapReduce的思想来将任务分步执行，这样的好处是更利于任务的分割与合并。现在描述可能没有多大的感觉，我们直接来看下面两个对比。
1. 基本流程梳理
按照常规思路，我们希望最终的结果是以Map形式存储，每个key存储单词，对应的value存储统计数量。于是，我们定义一个Map<String, Integer>类型用来存储最终的结果。数据集先使用一个String[]来代替，在最后的MR完整实现中，会从文件中进行读取。
static String[] text = { "what day is today", "today is a good day", "good good study", "day day up" };
2. 常规思路实现
如果只是单个的Java程序，我们可以这样做：
public static void main(String[] args) { // 定义用于存放统计结果的Map结构 Map<String, Integer> map = new HashMap<>(); // 读取数组中的每个元素，模拟一次读取一行 for (String line : text){ // 将每个单词以空格分割 String[] words = line.split(" "); // 读取每一个单词 for (String word : words){ // 每次将单词的统计结果取出，加1后放回 if (map.containsKey(word)){ map.put(word, map.get(word) + 1); }else { // 如果是第一次遇到这个单词，则存放1 map.put(word, 1); } } } // 输出结果 System.out.println(map); }
由于是简单的Java程序，这里就不过多说明了，大家可以自己看一下注释。
3. MR思想实现
从上面的程序可以看到，我们使用循环结构，逐行逐个的处理每行字符串中的每个单词，然后将结果不断的更新到Map结构中。在这种情况下，如果我们让不同的线程【相当于不同的Hadoop节点】去处理不同行的数据，再放到Map中时，为了考虑线程安全问题，其实是无法发挥最大作用的，很多时候要等待锁的释放。如果我们用MapReduce的思想来将程序改写一些就会不同了。
- 定义一个K-V键值对结构
static class KeyValuePair<K,V>{ K key; V value; public KeyValuePair(K key, V value){ this.key = key; this.value = value; } @Override public String toString() { return "{" + "key=" + key + ", value=" + value + '}'; } }
以下程序的编写可以帮助大家理解MR过程中最为重要的3个核心步骤：Map、Shuffling、Reduce。这三个阶段会完成许许多多的工作，对于开发者来说我们最关心的是数据结构上的变化，因此，其中涉及到的排序等相关操作并没有去实现，想要深挖的小伙伴可以去看源码。
- Map阶段
在这一阶段，会对数据逐行处理，key为偏移量，value则是这一行出现的数据键值对列表。
static Map<Integer, List<KeyValuePair<String, Integer>>> doMapper(){ Map<Integer, List<KeyValuePair<String, Integer>>> mapper = new HashMap<>(); // 定义偏移量指标，作为key int offset = 0; for (String line : text){ String[] words = line.split(" "); List<KeyValuePair<String, Integer>> list = new ArrayList<>(); for (String word : words){ // 将出现的单词作为键值对的key，将出现次数作为键值对的value KeyValuePair<String, Integer> keyValuePair = new KeyValuePair<>(word, 1); list.add(keyValuePair); } // 每次处理一行的数据，生成对应的键值对列表 mapper.put(offset, list); // 调整偏移量，总字符加一个换行符 offset += line.length() + 1; } return mapper; }
结果如下所示：
{0=[{key=what, value=1}, {key=day, value=1}, {key=is, value=1}, {key=today, value=1}], 18=[{key=today, value=1}, {key=is, value=1}, {key=a, value=1}, {key=good, value=1}, {key=day, value=1}], 38=[{key=good, value=1}, {key=good, value=1}, {key=study, value=1}], 54=[{key=day, value=1}, {key=day, value=1}, {key=up, value=1}]}
- Shuffling阶段
在这一阶段，将会把所有的key进行排序，并把相同的value放在同一个列表中。
static Map<String, List<Integer>> doShuffle(Map<Integer, List<KeyValuePair<String, Integer>>> mapper){ Map<String, List<Integer>> shuffle = new HashMap<>(); for (Integer key : mapper.keySet()){ List<KeyValuePair<String, Integer>> keyValuePairs = mapper.get(key); for (KeyValuePair<String, Integer> keyValuePair : keyValuePairs){ // 将出现过的相同单词放在同一个列表中 if (shuffle.containsKey(keyValuePair.key)){ shuffle.get(keyValuePair.key).add(keyValuePair.value); } else { // 如果是第一次记录，则创建一个列表 List<Integer> list = new ArrayList<>(); list.add(keyValuePair.value); shuffle.put(keyValuePair.key, list); } } } return shuffle; }
此时，依然不涉及计算逻辑，结果如图所示：
{a=[1], study=[1], what=[1], today=[1, 1], is=[1, 1], up=[1], day=[1, 1, 1, 1], good=[1, 1, 1]}
- Reduce阶段
在这一阶段，会在每个key对应的value列表中执行我们需要的计算逻辑。
static Map<String, Integer> doReducer(Map<String, List<Integer>> shuffle){ Map<String, Integer> reducer = new HashMap<>(); for (String key : shuffle.keySet()){ List<Integer> values = shuffle.get(key); Integer result = 0; // 此处对value进行处理，执行累加 for (Integer value : values){ result += value; } reducer.put(key, result); } return reducer; }
得到最终结果，执行结果如下：
{a=1, study=1, what=1, today=2, is=2, up=1, day=4, good=3}
- 程序运行结果
三、MapReduce
上面的例子帮大家简单的梳理了一下整体流程，这样我们就不需要debug去看每一步的执行效果了，因为只是模拟实现，所以省略了一些步骤。上面定义的KeyValuePair中出现的泛型也是整个流程的重要组成部分，实际执行计算任务时经常要根据需要合理的去定义Key与Value的类型。
1. Mapper
新建一个Class，继承Mapper，重写其中的map方法。可以先定义好泛型，然后再自动生成map方法。
import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.util.StringUtils; import java.io.IOException; /** * 以下泛型声明的是map阶段输入和输出数据的对应类型 * KEYIN: 偏移量，为整数类型 * VALUEIN: 每一行的字符串，为文本类型 * KEYOUT: 单词，为文本类型 * VALUEOUT: 出现次数1，为整数类型 */ public class WordCountMapper extends Mapper<LongWritable, Text, Text, LongWritable> { /** * map阶段将字符的偏移量作为key，每次得到的value为一行的数据 * @param key 字符偏移量，包含换行符 * @param value 整行的数据 * @param context 将结果输出到下一阶段的对象 */ @Override protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException { if (value != null){ // 获取该行的数据 String line = value.toString(); // 根据空格分离出每个单词 String[] words = StringUtils.split(line, ' '); // 将每个单词以键值对输出 for(String word : words){ context.write(new Text(word), new LongWritable(1)); } } } }
2. Reducer
新建一个Class，继承Reducer，重写其中的reduce方法。可以先定义好泛型，然后再自动生成reduce方法。
import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; /** * 以下泛型声明的是reduce阶段输入和输出数据的对应类型,输入类型对应的是Map阶段的输出 * KEYIN: 单词，为文本类型 * VALUEIN: 出现次数1，为整数类型 * KEYOUT: 单词，为文本类型 * VALUEOUT: 统计次数，为整数类型 */ public class WordCountReducer extends Reducer<Text, LongWritable, Text, LongWritable> { /** * 本例中省略了对shuffle的自定义，获取到的是默认处理后的数据 * @param key 单词 * @param values 出现1次的数据列表[1,1,...] * @param context 将结果最终输出的对象 */ @Override protected void reduce(Text key, Iterable<LongWritable> values, Reducer<Text, LongWritable, Text, LongWritable>.Context context) throws IOException, InterruptedException { // 定义用于记录累加结果的变量 long sum = 0; // 遍历列表，执行累加操作 for (LongWritable value : values){ sum += value.get(); } // 输出最后的统计结果 context.write(key, new LongWritable(sum)); } }
3. Executor
新建一个Class，继承Configured，并实现Tool接口，完整代码如下：
import edu.sand.mapper.WordCountMapper; import edu.sand.reducer.WordCountReducer; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; public class WordCountExecutor extends Configured implements Tool { @Override public int run(String[] strings) throws Exception { // 初始化配置，可以通过这个对象设置各种参数 Configuration conf = new Configuration(); // 完成Job初始化，设置任务名称 Job job = Job.getInstance(conf, "wordCount"); // 设置Job的运行主类 job.setJarByClass(WordCountExecutor.class); // 设置Map阶段的执行类 job.setMapperClass(WordCountMapper.class); // 设置Map阶段的数据输出类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(LongWritable.class); // 设置Reduce阶段的执行类 job.setReducerClass(WordCountReducer.class); // 设置Reduce阶段的数据输出类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(LongWritable.class); // 指定数据输入文件路径，如果指定的是文件夹，将读取目录下所有文件 FileInputFormat.setInputPaths(job, new Path("input/")); // 指定结果输出文件路径，最后一级路径会自动创建，每次重新执行时需要删除或修改名称 FileOutputFormat.setOutputPath(job, new Path("output/wordCount")); // 使用job调用执行，true代表显示详细信息，成功时返回0 return job.waitForCompletion(true) ? 0 : -1; } public static void main(String[] args) throws Exception { // 调用执行 ToolRunner.run(new Configuration(), new WordCountExecutor(), args); } }
4. 运行结果
- 项目结构说明
由于是本地代码运行，所以数据输入和结果输出都保存在本地磁盘上，可以在src同级创建两个文件夹input和output。
- 日志配置
如果希望看到更详细的日志输出，可以在resources文件夹下创建一个log4j.properties，内容如下：
log4j.rootLogger=INFO,stdout log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%p\t%d{ISO8601}\t%r\t%c\t[%t]\t%m%n
第一行的日志级别可以设置为INOF或者DEBUG。
- 执行结果
运行后会在对应路径下自动生成一个文件夹，其中主要包含3类文件：任务执行标志文件、结果输出文件、校验文件。以crc结尾的文件为校验类文件，当任务成功执行时，会产生一个**_SUCCESS文件，具体的运行结果会存放在part-r-xxxxx**文件中，part文件的名称和个数取决于Reduce的数量以及开发者的需要。
