[Hadoop in Action] 第4章 编写MapReduce基础程序
- 基于hadoop的专利数据处理示例
- MapReduce程序框架
- 用于计数统计的MapReduce基础程序
- 支持用脚本语言编写MapReduce程序的hadoop流式API
- 用于提升性能的Combiner
1、获取专利数据集
获取网址：http://www.nber.org/patents/
使用数据集：cite75_99.txt和apat63_99.txt
2、构建MapReduce程序的基础模版
代码清单 典型hadoop程序模版
import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.KeyValueTextInputFormat;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
public class MyJob extends Configured implements Tool {
public static class MapClass extends MapReduceBase
implements Mapper<Text, Text, Text, Text> {
public void map(Text key, Text value,
OutputCollector<Text, Text> output,
Reporter reporter) throws IOException {
output.collect(value, key);
}
}
public static class Reduce extends MapReduceBase
implements Reducer<Text, Text, Text, Text> {
public void reduce(Text key, Iterator<Text> values,
OutputCollector<Text, Text> output,
Reporter reporter) throws IOException {
String csv = "";
while (values.hasNext()) {
if (csv.length() > 0) csv += ",";
csv += values.next().toString();
}
output.collect(key, new Text(csv));
}
}
public int run(String[] args) throws Exception {
Configuration conf = getConf();
JobConf job = new JobConf(conf, MyJob.class);
Path in = new Path(args[0]);
Path out = new Path(args[1]);
FileInputFormat.setInputPaths(job, in);
FileOutputFormat.setOutputPath(job, out);
job.setJobName("MyJob");
job.setMapperClass(MapClass.class);
job.setReducerClass(Reduce.class);
job.setInputFormat(KeyValueTextInputFormat.class);
job.setOutputFormat(TextOutputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
job.set("key.value.separator.in.input.line", ",");
JobClient.runJob(job);
return 0;
}
public static void main(String[] args) throws Exception {
int res = ToolRunner.run(new Configuration(), new MyJob(), args);
System.exit(res);
}
}
框架的核心在run()方法中，也称为driver。它实例化、配置并传递一个JobConf对象命名的作业给JobClient.runJob()以启动MapReduce作业。JobConf对象将保持作业运行所需的全部配置参数。Driver需要在作业中为每个作业定制基本参数，包括输入路径、输出路径、Mapper类和Reducer类。另外，每个作业可以重置默认的作业属性，例如InputFormat、OutputFormat等，也可以调用JobConf对象中的set()方法填充人意配置参数。一旦传递JobConf对象到JobClient.runJob()，它就被视为作业的总体规划，成为决定这个作业如何运作的蓝本。
JobConf对象有许多参数，但我们并不希望全部的参数都通过编写driver来设置，可以把Hadoop安装时的配置文件作为一个很好的起点。同时，用户可能希望在命令行启动一个作业时传递额外的参数来改变作业配置。Driver可以通过自定义一组命令并自行处理用户参数，来支持用户修改其中的一些配置。因为经常需要做这样的任务，hadoop框架便提供了ToolRunner、Tool和Configured来简化其实现。
通过使用ToolRunner，MyJob可以自动支持下表中的选项：
|
选项
|
描述
|
| -conf <configuration file> | 指定一个配置文件 |
| -D <property=value> | 给JobConf属性赋值 |
| -fs <local | namenode:port> | 指定一个NameNode，可以是“local” |
| -jt <local | jobtracker:port> | 指定一个JobTracker |
| -files <list of files> | 指定一个以逗号分隔的文件列表，用于MapReduce作业。这些文件自动地分布到所有节点，使之可从本地获取 |
| -libjars <list of jars> | 指定一个以逗号分隔的jar文件列表，使之包含在所有任务JVM的classpath中 |
| -archives <list of archives> | 指定一个以逗号分隔的存档文件列表，使之可以在所有任务节点上打开 |
Mapper类和Reducer类如下所示：
public static class MapClass extends MapReduceBase
implements Mapper<k1, v1, k2, v2> {
public void map(k1 key, v1 value, OutputCollector<k2, v2> output, Reporter reporter) throws IOException { }
}
public static class Reducer extends MapReduceBase
implements Mapper<k2, v2, k3, v3> {
public void map(k2 key, Iterator<v2> values, OutputCollector<k3, v3> output, Reporter reporter) throws IOException { }
}
Mapper类的核心操作为map()方法，Reducer类为reduce()方法。每个map()方法的调用分别被赋予一个类型为k1和v1的键/值对。这个键/值对由mapper生成，并通过OutputCollector对象的collect()方法来输出。你需要在map()方法中的合适位置调用：
output.collect((k2) k, (v2) v);
在Reudcer中reduce()方法的每次调用均被赋予k2类型的键，以及v2类型的一组值。注意它必须与mapper中使用的k2和v2类型相同。Reduce()方法可能会循环遍历v2类型的所有值：
while (values.hasNext()) {
v2 v = values.next();
...
}
Reduce()方法还使用OutputCollector来搜集其键/值的输出，它们的类型为k3/v3。在reduce()方法中可以调用：
output.collect((k3) k, (v3) v);
除了在Mapper和Reducer之间保持k2和v2的类型一致，还需要确保在Mapper和Reducer中使用的键值类型与在driver中设置的输入格式、输出键的类，以及输出值的类保持一致。
3、计数
可以修改获取反向引用索引的程序来输出技术结果，只需要修改Reducer。如果选择让计数结果输出的类型为IntWritable，就需要在Reducer代码中的3个地方进行声明：
public static class Reduce extends MapReduceBase
implements Reducer<Text, Text, Text, IntWritable> {
public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, IntWritable> output,
Reporter reporter) throws IOException {
int count = 0;
while (values.hasNext()) {
values.next();
count++;
}
output.collect(key, new IntWritable(count));
}
}
- 编写MapReduce程序的第一步是了解数据流；
- 基于对数据流的理解，可以为输入、中间结果、输出的键/值对k1、v1、k2、v2、k3和v3设定类型；
- 根据数据流河数据类型，很容易能够理解程序代码。
代码清单 CitationHistogram.java
import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.KeyValueTextInputFormat;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
public class CitationHistogram extends Configured implements Tool {
public static class MapClass extends MapReduceBase
implements Mapper<Text, Text, IntWritable, IntWritable> {
private final static IntWritable uno = new IntWritable(1);
private IntWritable citationCount = new IntWritable();
public void map(Text key, Text value,
OutputCollector<IntWritable, IntWritable> output,
Reporter reporter) throws IOException {
citationCount.set(Integer.parseInt(value.toString()));
output.collect(citationCount, uno);
}
}
public static class Reduce extends MapReduceBase
implements Reducer<IntWritable,IntWritable,IntWritable,IntWritable>
{
public void reduce(IntWritable key, Iterator<IntWritable> values,
OutputCollector<IntWritable, IntWritable>output,
Reporter reporter) throws IOException {
int count = 0;
while (values.hasNext()) {
count += values.next().get();
}
output.collect(key, new IntWritable(count));
}
}
public int run(String[] args) throws Exception {
Configuration conf = getConf();
JobConf job = new JobConf(conf, CitationHistogram.class);
Path in = new Path(args[0]);
Path out = new Path(args[1]);
FileInputFormat.setInputPaths(job, in);
FileOutputFormat.setOutputPath(job, out);
job.setJobName("CitationHistogram");
job.setMapperClass(MapClass.class);
job.setReducerClass(Reduce.class);
job.setInputFormat(KeyValueTextInputFormat.class);
job.setOutputFormat(TextOutputFormat.class);
job.setOutputKeyClass(IntWritable.class);
job.setOutputValueClass(IntWritable.class);
JobClient.runJob(job);
return 0;
}
public static void main(String[] args) throws Exception {
int res = ToolRunner.run(new Configuration(),
new CitationHistogram(),
args);
System.exit(res);
}
}
4、适应Hadoop API的改变
（1）首先值得注意的是，在新的API中org.apache.hadoop.mapred的许多类都被移走了。多数被放入org.apache.hadoop.mapreduce,而且类库都放在org.apache.hadoop.mapreduce.lib的一个包中。当转为使用新API时，org.apache.hadoop.mapred下所有类的import声明就不存在了，它们都被弃用。
（2）新API中最有益的变化是引入了上下文对象context。最直接的影响在于替换了map()和reduce()方法中使用的OutputCollector和Reporter对象。深远的影响是统一了应用代码和MapReduce框架之间的通信，并固定了Mapper和Reduce的API，使得添加新功能时不会改变基本方法签名。
（3）新的map()和reduce()方法分别被包含在新的抽象类Mapper和Reducer中。它们取代了原始API中的Mapper和Reducer接口。新的抽象类也替换了MapReduceBase类，使之被弃用。
（4）新的map()和Reduce()方法多了一两处细微的改变。它们可以抛出InterruptedException而非单一的IOException。而且，reduce()方法不再以Iterator而以Iterable来接受一个值的列表，这样更容易使用Java的foreach语义来实现迭代。
原始API中的签名
public static class MapClass extends MapReduceBase
implements Mapper<k1, v1, k2, v2> {
public void map(k1 key, v1 value, OutputCollector<k2, v2> output, Reporter reporter) throws IOException { }
}
public static class Reducer extends MapReduceBase
implements Mapper<k2, v2, k3, v3> {
public void map(k2 key, Iterator<v2> values, OutputCollector<k3, v3> output, Reporter reporter) throws IOException { }
}
新API一定程度上对它们做了简化
public static class MapClass extends Mapper<k1, v1, k2, v2> {
public void map(k1 key, v1 value, Context context) throws IOException, InterruptedException { }
}
public static class Reduce extends Reducer<k2, v2, k3, v3> {
public void map(k2 key, Iterable<v2> value, Context context) throws IOException, InterruptedException { }
}
（5）还需要改变driver中的一些内容来支持新的API。在新的API中JobConf和JobClient被替换了。它们的功能已经被放入Configuration类和一个新的类Job中。Configuration类纯粹为了配置作业而设，而Job类负责定义和控制一个作业的执行。作业的构造和提交执行现在放在Job中。
原API
JobConf job = new JobConf(conf, MyJob.calss);
job.setJobName(“MyJob");
JobClient.runJob(job);
新API
Job job = new Job(conf, “MyJob”);
job.setJarByClass(MyJob.class);
System.exit(job.waitForCompletion(true)?0:1);
代码清单 基于版本0.20新API重写的hadoop基础程序模版
import java.io.IOException;
import java.util.Iterable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;
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.util.Tool;
import org.apache.hadoop.util.ToolRunner;
public class MyJob extends Configured implements Tool {
public static class MapClass extends Mapper<LongWritable, Text, Text, Text> {
public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String[] citation = value.toString().split(",");
context.write(new Text(citation[1]), new Text(citation[0]));
}
}
public static class Reduce extends Reducer<Text, Text, Text, Text> {
public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
String csv = "";
for (Text val:values) { //Iterable类型允许foreach循环
if (csv.length() > 0) csv += ",";
csv += val.toString();
}
context.write(key, new Text(csv));
}
}
public int run(String[] args) throws Exception {
Configuration conf = getConf();
Job job = new Job(conf, "MyJob");
job.setJarByClass(MyJob.class);
Path in = new Path(args[0]);
Path out = new Path(args[1]);
FileInputFormat.setInputPaths(job, in);
FileOutputFormat.setOutputPath(job, out);
job.setMapperClass(MapClass.class);
job.setReducerClass(Reduce.class);
job.setInputFormatClass(TextInputFormat.class); //兼容的InputFormat类
job.setOutputFormatClass(TextOutputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
System.exit(job.waitForCompletion(true)?0:1);
return 0;
}
public static void main(String[] args) throws Exception {
int res = ToolRunner.run(new Configuration(), new MyJob(), args);
System.exit(res);
}
}
5、Hadoop的Streaming
- 通过Unix命令使用Streaming
- 通过脚本使用Streaming
- 用Streaming处理键/值对
- 通过Aggregate包使用Streaming
6、使用combiner提升性能
hadoop通过扩展MapReduce框架，在mapper和reducer之间增加了一个combiner。你可以将combiner视为reducer的助手。它致力于减少mapper的输出以降低网络和reducer上的压力。为了使combiner工作，它在数据的转换上必须与reducer等价。
[转载请注明] http://www.cnblogs.com/zhengrunjian/
[Hadoop in Action] 第4章 编写MapReduce基础程序的更多相关文章
- 第 3 章 编写 PAM 应用程序和服务
Solaris 开发者安全性指南 Previous: 第 2 章 开发特权应用程序 Next: 第 4 章 编写使用 GSS-API 的应用程序 第 3 章 编写 PAM 应用程序和服务 可插拔验证模 ...
- [Hadoop in Action] 第5章 高阶MapReduce
链接多个MapReduce作业 执行多个数据集的联结 生成Bloom filter 1.链接MapReduce作业 [顺序链接MapReduce作业] mapreduce-1 | mapr ...
- [Hadoop in Action] 第7章 细则手册
向任务传递定制参数 获取任务待定的信息 生成多个输出 与关系数据库交互 让输出做全局排序 1.向任务传递作业定制的参数 在编写Mapper和Reducer时,通常会想让一些地方可以配 ...
- [Hadoop in Action] 第1章 Hadoop简介
编写可扩展.分布式的数据密集型程序和基础知识 理解Hadoop和MapReduce 编写和运行一个基本的MapReduce程序 1.什么是Hadoop Hadoop是一个开源的框架,可编写和运 ...
- [Hadoop in Action] 第6章 编程实践
Hadoop程序开发的独门绝技 在本地,伪分布和全分布模式下调试程序 程序输出的完整性检查和回归测试 日志和监控 性能调优 1.开发MapReduce程序 [本地模式] 本地模式 ...
- [hadoop in Action] 第3章 Hadoop组件
管理HDFS中的文件 分析MapReduce框架中的组件 读写输入输出数据 1.HDFS文件操作 ［命令行方式］ Hadoop的文件命令采取的形式为: hadoop fs -cmd < ...
- [Hadoop in Action] 第2章 初识Hadoop
Hadoop的结构组成 安装Hadoop及其3种工作模式:单机.伪分布和全分布 用于监控Hadoop安装的Web工具 1.Hadoop的构造模块 (1)NameNode(名字节点) ...
- 《Hadoop权威》学习笔记五：MapReduce应用程序
一.API的配置---Configuration类 API的配置:Hadoop提供了专门的API对资源进行配置,Configuration类的实例(在org.apache.hadoop.conf包)包 ...
- Hadoop：使用Mrjob框架编写MapReduce
Mrjob简介 Mrjob是一个编写MapReduce任务的开源Python框架,它实际上对Hadoop Streaming的命令行进行了封装,因此接粗不到Hadoop的数据流命令行,使我们可以更轻松 ...
随机推荐
- Java集合框架实现自定义排序
Java集合框架针对不同的数据结构提供了多种排序的方法,虽然很多时候我们可以自己实现排序,比如数组等,但是灵活的使用JDK提供的排序方法,可以提高开发效率,而且通常JDK的实现要比自己造的轮子性能更优 ...
- Linux系统启动级别及顺序
Linux系统有7个运行级别(runlevel)运行级别0:系统停机状态,系统默认运行级别不能设为0,否则不能正常启动运行级别1:单用户工作状态,root权限,用于系统维护,禁止远程登陆运行级别2:多 ...
- 首届Ignite China微软技术大会见闻
10.26-10.28,有幸参加微软在中国北京举办的首届Ignite China技术大会.世界那么大,技术那么多,我想去看看. 为期三天的技术大会在小汤山九华山庄举办,吐槽一下,太特么远了,每天要跑3 ...
- soapui中文操作手册(一)----创建一个新的项目
1) 创建一个新的项目 点击项目,选择新建项目SOAP.这将打开一个新的SOAP项目对话框. 注意:你也可以做CTRL + N(WIN)或CMD+ N(MAC)来创建一个新的SOAP项目. 在新的SO ...
- vim molokai配色方案（调过）
" Vim color file " " Author: Tomas Restrepo <tomas@winterdom.com> " " ...
- 关于 iOS 批量打包的总结
关于 iOS 批量打包的总结 本文作者: 伯乐在线 - Tsui YuenHong .未经作者许可,禁止转载!欢迎加入伯乐在线 专栏作者. 如果你曾经试过做多 target 的项目,到了测试人员要 ...
- Android -- ViewRoot，关于子线程刷新UI
Android在4.0之后执行线程更新UI操作会报异常:CalledFromWrongThreadException:Only the original thread that created a v ...
- 团队开发（NABC）
特点:这是一个手机软件,能通过通讯录录入生日信息 N(Need需求):现在在交际圈中需要记住越来越多朋友的生日信息 A(Approach做法):由一个简单的闹钟为基础,添加与生日相关的功能,最终实现 ...
- 面向对象版js分页
基于前一个js分页,我将代码改成一个面向对象版的js分页,代码如下 http://peng666.github.io/blogs/pageobj <!DOCTYPE html> <h ...
- Combox和DropDownList控件的区别
共同点:都是下拉框控件 不同点:Combox用在winform上,DropDownList用在网页上,且两者绑定方式略有不同 绑定数据例子如下—— 1.Combox绑定 DataTable dtBus ...