MapReduce设计模式：Chaining
MapReduce设计模式：Chaining
?
Chaining这种设计模式非常重要，主要是因为你通常无法通过单个MapReduce Job来完成工作；某些Job必须串行，因为前者Job的输出会成为下个Job的输入；某些Job可以并行，因为Job运行之间没有关系；有些Job的Mapper是对日志的重复处理，需要将代码规范化。不管怎么说，不能靠单个MapReduce程序就完成工作是我们的挑战。
?
我们这节来讨论下Chaining模式，由于Hadoop的设计是为了MapReduce程序的高速运行，单个Job完全没有问题，但是这个并不能解决工作中的问题。在 工作中，通常任务根本无法通过单个Job完成，除非是像WordCount这样的任务。也就是说对于多阶段的MapReduce程序控制上需要我们做更多的编码，对Job进行控制。如果你感觉任务过于复杂，控制难于从心时，看下这个Oozie这个Apache Project，这个Oozie我也没有细看，能够通过配置对Job进行更为细致的控制。我们本节的任务不是讨论MapReduce框架处理Job的复杂性，不过对于Job执行的优劣必须知道。
?
下面我们讨论Chaining设计模式的实现。
首先，回忆下我们讨论过的TOP-K问题，会怎么实现呢？
?
首先我们先对输出的数据进行排序，算出总排序，这个需要一个Job；然后再对这个Job的输出作为下一个Job的输入，计算TOP-K数据，代码不再单独列出，思想理解就行。
这个计算方法就利用了一种很重要的Chaining方法，Serial Chaining。这个是指将MapReduce Job逐个串起来，按照顺序执行。用Unix-Style方式表示就是：
mapreduce-1 | mapreduce-2 | mapreduce-3 | ...
?这个是最直观的Mapreduce Job的编写方式。通过这种控制，我们能完成工作，但是问题在于：这样写出来的代码如果Job比较多的话，控制各个mapreduce的执行状态会让代码特别丑；如果只有少数几个的话，还是可以接受的，如果Job串行过多，这种方法要慎用。
其实Hadoop的Mapreduce支持依赖方式的编码。如果有这么个场景：mapreducea处理来自数据库的数据集合，mapreduceb处理来自Hbase的数据集合，mapreducec则对这两个集合进行Join操作，Join操作的内容参考http://isilic.iteye.com/blog/1772945。这样这三个mapreduce就有了依赖关系。mapreducea和mapreduceb两者可以并行执行，mapreducec则依赖于前两者的执行结果。这个用上面提到的Serial Chaining也不好解决。
实际上，Hadoop实现的mapreduce程序已经帮助我们完成相关的控制；在Job中，可以设置依赖的Job，直接提交最终的Job，MapReduce会控制依赖关系的执行：
mapreducec.addDependingJob(mapreducea);mapreducec.addDependingJob(mapreduceb);submit(mapreducea,mapreduceb,mapreducec);
?
注意这个并不是真正的mapreduce依赖，是Job的依赖，这里只是点明三者的关系。
使用这种方法，需要对Job的执行状态进行监控，确保三者都执行完毕后，主程序再退出。
?
上面两种方法都是对于Job依赖的情况，前者是自己控制，后者定义好依赖关系，提交给系统控制。不过现在又这么一种情况，在Document Information Retrieval中，需要对文档的Stop word进行处理，还需要对Stemming进行继续的处理，后续再进行lemmatization处理，处理完成后作为Term进行后续的正常处理，如词频统计，分词等。对于Document的一系列操作难道要写多个Mapreduce吗？不用，Hadoop提供了ChainMapper和ChainReducer来处理这种情况。我们来详细学习下ChainMapper和ChainReducer这两个方法。
?
ChainMapper：在单个map节点执行多个mapper逻辑，这个类似于Unix的管道操作，管道处理记录；其中前者的处理输出作为下一个的mapper的输入，mapper chain执行完毕后，才会进入到partitioner阶段，后面会继续进入到Reducer阶段。
?
ChainReducer：难道你会认为ChainReducer的功能和ChainMapper类似吗？No，完全不是。ChainReducer中可以设置单个Reducer和多个Mapper，执行完Reducer之后会再顺序执行Mapper的逻辑（吐槽下这个ChainReducer的命名），跟Reducer Chain完全没有关系。
?
按照官网的说法，ChainMapper的优点是更好的利用IO，在大多数情况下，单个Mapper也能完成ChainMapper的内容，不过这样的设计更好的是软件工程的功能化、内聚化的体现。ChainReducer能偶更好地利用reducer节点，方便实现reducer输出后的进一步数据逻辑处理，实际上就是ChainReducer中的Mapper处理逻辑。
本文开始的mapreduce程序可以逻辑表述为：
?
[ map | reduce]+
?
ChainMapper和ChainReducer用逻辑关系式可以表示为：
[ map* | reducer mapper* ]
?这个逻辑表达式一定要理解数据流的走向，最好能结合代码看下。
?
将上面的内容结合起来，就可以得到本节的核心内容，表示为：
[ map* | reducer mapper* ]+
这个关系式如果能理解，再结合自己工作中的逻辑，就算是深入的理解本节的内容了。
?
最后贴个wordcount的代码，根据本节的内容做了个更改，逻辑性方面就不要再追究了，仅供演示使用。
代码不做解释，大家就是需要理解一下这个过程：
?
public class WordCount { public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> { private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map ( Object key, Text value, Context context ) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one); } } } public static class UpperCaseMapper extends Mapper<Text, IntWritable, Text, IntWritable> { public void map ( Text key, IntWritable value, Context context ) throws IOException, InterruptedException { String k = key.toString(); context.write(new Text(k.toUpperCase()), value); } } public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private IntWritable result = new IntWritable(); public void reduce ( Text key, Iterable<IntWritable> values, Context context ) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } } public static class ThresholdFilterMapper extends Mapper<Text, IntWritable, Text, IntWritable> { public void map ( Text key, IntWritable value, Context context ) throws IOException, InterruptedException { int cnt = value.get(); if (cnt > 10) context.write(key, value); } } 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> <out>"); System.exit(2); } JobConf jc = new JobConf("JobControl"); Job job = new Job(jc); Configuration tokenConf = new Configuration(false); ChainMapper.addMapper(job, TokenizerMapper.class, Object.class, Text.class, Text.class, IntWritable.class, tokenConf); Configuration upperConf = new Configuration(false); ChainMapper.addMapper(job, UpperCaseMapper.class, Text.class, IntWritable.class, Text.class, IntWritable.class, upperConf); job.setJarByClass(WordCount.class); job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); Configuration reducerConf = new Configuration(false); ChainReducer.setReducer(job, IntSumReducer.class, Text.class, IntWritable.class, Text.class, IntWritable.class, reducerConf); Configuration thresholdConf = new Configuration(false); ChainReducer.addMapper(job, ThresholdFilterMapper.class, Text.class, IntWritable.class, Text.class, IntWritable.class, thresholdConf); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0])); FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); }}?