如何将一个反向引用索引的程序的Reducer输出的类型改为IntWritable
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));
}
}
计算不同引用次数专利的数目
之所以选择K2、V2、K3和V3的数据为IntWritable类型，是因为它们的数据必然为整数，并且使用IntWritable比Text更高效
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.setInputFomrat(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);
}
}
需要说明的几点：
1、类名为CitationHistogram
2、输入格式为KeyValueTextInputFormat，输出格式为TextOutputFormat
3、KeyValueTextInputFormat默认以"<tab>"制表符进行分割，可以使用job.set("key.value.separator.in.input.line", ",")，修改分隔符为","，其他需要自行修改
4、MapClass类使用了两个私有成员变量uno 和citationCount，为什么要这样定义呢？
出于对效率的考虑citationCount和uno的定义被放在类中而不是方法中，有多少记录，map()方法就会被调用多少次 (对每个JVM而言，就是一个分片中的记录数)。减少在map()方法中生成的对象个数可以提高性能，并减少垃圾回收。由于citationCount和uno被传递给output.collect()，我们依赖output.collect()方法的约定不会修改这两个对象。
5、Reducer计算每个key对应的值的总数，这似乎并不高效，因为我们知道所有的值都是1，为什么我们还要去加它们呢？原因在于，它会让我们在以后能更容易地增加一个combiner来提高性能。
Hadoop API是改了又改，变了又变
1、0.20版本被视为旧API和未来稳定API之间的过渡
2、为了保持向后兼容，版本0.20以及之后的版本支持下一代API并将旧API标记为弃用
并不推荐马上转向新的API
1、0.20版本，许多Hadoop自身的类库还没有基于新的API重写，如果MapReduce代码使用0.20中的新API，这些类将无法被使用。
2、在0.20版本之前，0.18.3仍被很多人认为是最完备与最稳定的版本
3、虽然在新版本的API中有所改变，但是改变的仅仅影响基础的MapReduce模板，我们可以基于新API所做的改变，重写这个模板以备将来使用。
你可能会奇怪为什么不提0.19？
一般的意见认为它的初始版本问题比较多，有许多bug，一些副版本试图解决这个问题，但社区希望直接跳到版本0.20
新版本的API做了哪些改动？
1、在新的API中org.apahce.hadoop.mapred的许多类都被移走了，多数被放入org.apache.hadoop.mapreduce中，而且类库都放在org.apache.hadoop.mapreduce.lib的一个包里。当转为使用新API时，org.apache.hadoop.mapred下所有的类的import声明(或者完全引用)就不存在了，它们都被弃用
2、新API中最有益的变化是引入了上下文对象context，最直接的影响在于替换了map()和reduce()方法中使用的OutputCollector和Reporter对象。现在将通过调用Context.writer()而不是OutputCollector.collect()输出键/值对。深远的影响是统一了应用代码和MapReduce框架之间的通信，并固定了Mapper和Reducer的API，使得添加新功能时不会改变基本方法签名，新的功能仅仅时在context对象上增加的方法，在引入这些方法之前写的程序不会感知到这些新方法，它们可以在更新的版本中继续编译与运行
3、新的map()和reduce()方法分别被包含在新的抽象Mapper和Reducer中，它们取代了原始API中的Mapper和Reducer接口(org.apache.hadoop.mapred.Mapper和org.apache.hadoop.mapred.Reducer)。新的抽象类也替换了MapReudceBase类，使之被弃用
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 Reduce extends MapReduceBase
implements Reducer<K2, V2, K3, V3> {
public void reduce(K2 key, Iterator<V2> values,
OutputCollector<K3, V3> output,
Reporter reporter) throws IOException { }
}
新的API中的签名，体会与上面签名的不同
public static class MapClass extends Mapper<K1, V2, 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 reduce(K2 key, Iterable<V2> values, Context context)
throws IOException, InterruptedException { }
}
你还需要改变driver中的一些内容来支持新的API
1、在新的API中JobConf和JobClient被替换了，它们的功能已经被放入Configuration类(它原本是JobConf的父类)和一个新的类Job中
2、Configuration类纯粹为了配置作业而设，而Job类负责定义和控制一个作业的执行
3、比如，setOutputKeyClass()和setOutputValueClass()等方法被从JobConf转移到了Job
4、作业的构造和提交执行现在放在Job中，原本需要使用JobConf来构造一个作业：
JobConf job = new JobConf(conf, MyJob.class);
job.setJobName("MyJob");
而现在可通过Job类完成：
Job job = new Job(conf, "MyJob");
job.setJarByClass(MyJob.class);
以前是通过JobClient提交作业去执行：
JobClient.runJob(job);
现在同样通过Job类来完成
System.exit(job.waitForCompletion(true) ? 0 : 1);
基于版本0.20及其以上的新API重写的Hadoop基础程序模板（Hadoop 1.X也适用）
public class MyJob extends Configured implements Tool {
public static class MapClass extends Mapper<LongWritable, Text, Text, Text> {
public void map(LongWirtable 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> {
// Iterable类型允许foreach循环
public void reduce(Text key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
for (Text val : values) {
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);
// 兼容InputFormat类
job.setInputFormatClass(TextInputFormat.class);
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);
}
}
1、这段代码实现了反向索引功能
2、KeyValueTextInputFormat类未被移植到版本0.20的新API中，重写这个模板我们不得不使用TextInputFormat