Hadoop学习笔记—4.初识MapReduce

一、神马是高大上的MapReduce

　　MapReduce是Google的一项重要技术，它首先是一个编程模型，用以进行大数据量的计算。对于大数据量的计算，通常采用的处理手法就是并行计算。但对许多开发者来说，自己完完全全实现一个并行计算程序难度太大，而MapReduce就是一种简化并行计算的编程模型，它使得那些没有多有多少并行计算经验的开发人员也可以开发并行应用程序。这也就是MapReduce的价值所在，通过简化编程模型，降低了开发并行应用的入门门槛。

1.1 MapReduce是什么

　　Hadoop MapReduce是一个软件框架，基于该框架能够容易地编写应用程序，这些应用程序能够运行在由上千个商用机器组成的大集群上，并以一种可靠的，具有容错能力的方式并行地处理上TB级别的海量数据集。这个定义里面有着这些关键词，一是软件框架，二是并行处理，三是可靠且容错，四是大规模集群，五是海量数据集。

1.2 MapReduce做什么

package mapreduce;

import java.io.FileInputStream;
import java.io.IOException;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;

public class MyWordCountJob {

/**
* @author Edison Chou
* @version 1.0
* @param KEYIN
*            →k1 表示每一行的起始位置（偏移量offset）
* @param VALUEIN
*            →v1 表示每一行的文本内容
* @param KEYOUT
*            →k2 表示每一行中的每个单词
* @param VALUEOUT
*            →v2 表示每一行中的每个单词的出现次数，固定值为1
*/
public static class MyMapper extends
Mapper<LongWritable, Text, Text, LongWritable> {
protected void map(LongWritable key, Text value,
Mapper<LongWritable, Text, Text, LongWritable>.Context context)
throws java.io.IOException, InterruptedException {
String[] spilted = value.toString().split(" ");
for (String word : spilted) {
context.write(new Text(word), new LongWritable(1L));
}
};
}

/**
* @author Edison Chou
* @version 1.0
* @param KEYIN
*            →k2 表示每一行中的每个单词
* @param VALUEIN
*            →v2 表示每一行中的每个单词的出现次数，固定值为1
* @param KEYOUT
*            →k3 表示每一行中的每个单词
* @param VALUEOUT
*            →v3 表示每一行中的每个单词的出现次数之和
*/
public static class MyReducer extends
Reducer<Text, LongWritable, Text, LongWritable> {
protected void reduce(Text key,
java.lang.Iterable<LongWritable> values,
Reducer<Text, LongWritable, Text, LongWritable>.Context context)
throws java.io.IOException, InterruptedException {
long count = 0L;
for (LongWritable value : values) {
count += value.get();
}
context.write(key, new LongWritable(count));
};
}

// 输入文件路径
public static final String INPUT_PATH = "hdfs://hadoop-master:9000/testdir/input/words.txt";
// 输出文件路径
public static final String OUTPUT_PATH = "hdfs://hadoop-master:9000/testdir/output/wordcount";

public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();

// 0.0:首先删除输出路径的已有生成文件
FileSystem fs = FileSystem.get(new URI(INPUT_PATH), conf);
Path outPath = new Path(OUTPUT_PATH);
if (fs.exists(outPath)) {
fs.delete(outPath, true);
}

Job job = new Job(conf, "WordCount");
job.setJarByClass(MyWordCountJob.class);

// 1.0:指定输入目录
FileInputFormat.setInputPaths(job, new Path(INPUT_PATH));
// 1.1:指定对输入数据进行格式化处理的类（可以省略）
job.setInputFormatClass(TextInputFormat.class);
// 1.2:指定自定义的Mapper类
job.setMapperClass(MyMapper.class);
// 1.3:指定map输出的<K,V>类型（如果<k3,v3>的类型与<k2,v2>的类型一致则可以省略）
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(LongWritable.class);
// 1.4:分区（可以省略）
job.setPartitionerClass(HashPartitioner.class);
// 1.5:设置要运行的Reducer的数量（可以省略）
job.setNumReduceTasks(1);
// 1.6:指定自定义的Reducer类
job.setReducerClass(MyReducer.class);
// 1.7:指定reduce输出的<K,V>类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(LongWritable.class);
// 1.8:指定输出目录
FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH));
// 1.9:指定对输出数据进行格式化处理的类（可以省略）
job.setOutputFormatClass(TextOutputFormat.class);
// 2.0:提交作业
boolean success = job.waitForCompletion(true);
if (success) {
System.out.println("Success");
System.exit(0);
} else {
System.out.println("Failed");
System.exit(1);
}
}

}

View Code

3.5 运行吧小DEMO

　　（1）调试查看控制台状态信息



　　（2）通过Shell命令查看统计结果

四、使用ToolRunner类改写WordCount

　　Hadoop有个ToolRunner类，它是个好东西，简单好用。无论在《Hadoop权威指南》还是Hadoop项目源码自带的example，都推荐使用ToolRunner。

4.1 最初的写法

　　下面我们看下src/example目录下WordCount.java文件，它的代码结构是这样的：

public class WordCount {
// 略...
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf,
args).getRemainingArgs();
// 略...
Job job = new Job(conf, "word count");
// 略...
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}

　　WordCount.java中使用到了GenericOptionsParser这个类，它的作用是将命令行中参数自动设置到变量conf中。举个例子，比如我希望通过命令行设置reduce task数量，就这么写：

bin/hadoop jar MyJob.jar com.xxx.MyJobDriver -Dmapred.reduce.tasks=5

　　上面这样就可以了，不需要将其硬编码到java代码中，很轻松就可以将参数与代码分离开。

4.2 加入ToolRunner的写法

　　至此，我们还没有说到ToolRunner，上面的代码我们使用了GenericOptionsParser帮我们解析命令行参数，编写ToolRunner的程序员更懒，它将 GenericOptionsParser调用隐藏到自身run方法，被自动执行了，修改后的代码变成了这样：

public class WordCount extends Configured implements Tool {
@Override
public int run(String[] arg0) throws Exception {
Job job = new Job(getConf(), "word count");
// 略...
System.exit(job.waitForCompletion(true) ? 0 : 1);
return 0;
}

public static void main(String[] args) throws Exception {
int res = ToolRunner.run(new Configuration(), new WordCount(), args);
System.exit(res);
}
}

　　看看这段代码上有什么不同：

　　（1）让WordCount继承Configured并实现Tool接口。

　　（2）重写Tool接口的run方法，run方法不是static类型，这很好。

　　（3）在WordCount中我们将通过getConf()获取Configuration对象。

　　可以看出，通过简单的几步，就可以实现代码与配置隔离、上传文件到DistributeCache等功能。修改MapReduce参数不需要修改java代码、打包、部署，提高工作效率。

4.3 重写WordCount程序

public class MyJob extends Configured implements Tool {
public static class MyMapper extends
Mapper<LongWritable, Text, Text, LongWritable> {
protected void map(LongWritable key, Text value,
Mapper<LongWritable, Text, Text, LongWritable>.Context context)
throws java.io.IOException, InterruptedException {
......
}
};
}

public static class MyReducer extends
Reducer<Text, LongWritable, Text, LongWritable> {
protected void reduce(Text key,
java.lang.Iterable<LongWritable> values,
Reducer<Text, LongWritable, Text, LongWritable>.Context context)
throws java.io.IOException, InterruptedException {
......
};
}

// 输入文件路径
public static final String INPUT_PATH = "hdfs://hadoop-master:9000/testdir/input/words.txt";
// 输出文件路径
public static final String OUTPUT_PATH = "hdfs://hadoop-master:9000/testdir/output/wordcount";

@Override
public int run(String[] args) throws Exception {
// 首先删除输出路径的已有生成文件
FileSystem fs = FileSystem.get(new URI(INPUT_PATH), getConf());
Path outPath = new Path(OUTPUT_PATH);
if (fs.exists(outPath)) {
fs.delete(outPath, true);
}

Job job = new Job(getConf(), "WordCount");
// 设置输入目录
FileInputFormat.setInputPaths(job, new Path(INPUT_PATH));
// 设置自定义Mapper
job.setMapperClass(MyMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(LongWritable.class);
// 设置自定义Reducer
job.setReducerClass(MyReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(LongWritable.class);
// 设置输出目录
FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH));

System.exit(job.waitForCompletion(true) ? 0 : 1);
return 0;
}

public static void main(String[] args) {
Configuration conf = new Configuration();
try {
int res = ToolRunner.run(conf, new MyJob(), args);
System.exit(res);
} catch (Exception e) {
e.printStackTrace();
}
}

}

参考资料

（1）王路情，《Hadoop之MapReduce》：/article/2766062.html

（2）Suddenly，《Hadoop日记之MapReduce》：/content/5045598.html

（3）伯乐在线，《我是如何向老婆解释MapReduce的》：http://blog.jobbole.com/1321/

（4）codingwu，《MapReduce原理与设计思想》：http://www.cnblogs.com/archimedes/p/mapreduce-principle.html

（5）codingwu，《MapReduce实例浅析》：http://www.cnblogs.com/archimedes/p/mapreduce-example-analysis.html

（6）挑灯看剑，《图解MapReduce原理和执行过程》：/article/1518437.html

（7）万川梅、谢正兰，《Hadoop应用开发实战详解（修订版）》：http://item.jd.com/11508248.html

（8）张月，《Hadoop MapReduce开发最佳实践》：http://www.infoq.com/cn/articles/MapReduce-Best-Practice-1

作者：周旭龙

出处：http://edisonchou.cnblogs.com/

本文版权归作者和博客园共有，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文链接。