Hadoop的ChainMapper和ChainReducer使用案例（链式处理）（四）

jiang1799

　　不多说，直接上干货！
　　Hadoop的MR作业支持链式处理，类似在一个生产牛奶的流水线上，每一个阶段都有特定的任务要处理，比如提供牛奶盒，装入牛奶，封盒，打印出厂日期，等等，通过这样进一步的分工，从而提高了生产效率，那么在我们的Hadoop的MapReduce中也是如此，支持链式的处理方式，这些Mapper像Linux管道一样，前一个Mapper的输出结果直接重定向到下一个Mapper的输入，形成一个流水线，而这一点与Lucene和Solr中的Filter机制是非常类似的，Hadoop项目源自Lucene，自然也借鉴了一些Lucene中的处理方式。
　　举个例子，比如处理文本中的一些禁用词，或者敏感词，等等，Hadoop里的链式操作，支持的形式类似正则Map+ Rrduce Map*，代表的意思是全局只能有一个唯一的Reduce，但是在Reduce的前后是可以存在无限多个Mapper来进行一些预处理或者善后工作的。
　　注意：
　　1. 本人目前使用的版本是1.2.1，因此ChainMapper使用的还是old api。
　　2. 老的API之中，只支持 N-Mapper + 1-Reducer的模式。 Reducer不在链式任务最开始即可。
　　比如：
　　Map1 -> Map2 -> Reducer -> Map3 -> Map4
　　(不确定在新版的API之中是否支持 N-Reducer的模式。不过new api 确实要简单简洁很多)

　　在编程的时候，我们可以借用源码提供给我们的程序！在此基础上进行修改和编写。
　　比如我的源码本地目录如下：（找我的本地ChainMapper和ChainReducer案例）
　　D:\SoftWare\hadoop-2.2.0-src\hadoop-mapreduce-project\hadoop-mapreduce-client\hadoop-mapreduce-client-core\src\main\java\org\apache\hadoop\mapreduce\lib\chain

　　任务介绍：
　　这个任务需要两步完成：
　　1. 对一篇文章进行WordCount
　　2. 统计出现次数超过5词的单词
　　WordCount我们很熟悉，因为版本限制，先使用old api 实现一次：
　　Java代码

• package hadoop_in_action_exersice;
  
• 
  
• import java.io.IOException;
  
• import java.util.Iterator;
  
• import java.util.StringTokenizer;
  
• 
  
• import org.apache.hadoop.fs.FileSystem;
  
• import org.apache.hadoop.fs.Path;
  
• import org.apache.hadoop.io.IntWritable;
  
• import org.apache.hadoop.io.LongWritable;
  
• 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.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.TextInputFormat;
  
• import org.apache.hadoop.mapred.TextOutputFormat;
  
• 
  
• public class ChainedJobs {
  
• 
  
•     public static class TokenizeMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {
  
• 
  
•         private final static IntWritable one = new IntWritable(1);
  
•         public static final int LOW_LIMIT = 5;
  
•         @Override
  
•         public void map(LongWritable key, Text value,
  
•                 OutputCollector<Text, IntWritable> output, Reporter reporter)
  
•                 throws IOException {
  
•             String line = value.toString();
  
•             StringTokenizer st = new StringTokenizer(line);
  
•             while(st.hasMoreTokens())
  
•                 output.collect(new Text(st.nextToken()), one);
  
• 
  
•         }
  
• 
  
•     }
  
• 
  
•     public static class TokenizeReducer extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {
  
• 
  
•         @Override
  
•         public void reduce(Text key, Iterator<IntWritable> values,
  
•                 OutputCollector<Text, IntWritable> output, Reporter reporter)
  
•                 throws IOException {
  
•             int sum = 0;
  
•             while(values.hasNext()) {
  
•                 sum += values.next().get();
  
•             }
  
•             output.collect(key, new IntWritable(sum));
  
•         }
  
• 
  
•     }
  
• 
  
• 
  
•     public static void main(String[] args) throws IOException {
  
• 
  
• 
  
•         JobConf conf = new JobConf(ChainedJobs.class);
  
•         conf.setJobName("wordcount");           //设置一个用户定义的job名称
  
•         conf.setOutputKeyClass(Text.class);    //为job的输出数据设置Key类
  
•         conf.setOutputValueClass(IntWritable.class);   //为job输出设置value类
  
•         conf.setMapperClass(TokenizeMapper.class);         //为job设置Mapper类
  
•         conf.setCombinerClass(TokenizeReducer.class);      //为job设置Combiner类
  
•         conf.setReducerClass(TokenizeReducer.class);        //为job设置Reduce类
  
•         conf.setInputFormat(TextInputFormat.class);    //为map-reduce任务设置InputFormat实现类
  
•         conf.setOutputFormat(TextOutputFormat.class);  //为map-reduce任务设置OutputFormat实现类
  
• 
  
•         // Remove output folder before run job(s)
  
•         FileSystem fs=FileSystem.get(conf);
  
•         String outputPath = "/home/hadoop/DataSet/Hadoop/WordCount-OUTPUT";
  
•         Path op=new Path(outputPath);
  
•         if (fs.exists(op)) {
  
•             fs.delete(op, true);
  
•             System.out.println("存在此输出路径，已删除！！！");
  
•         }
  
• 
  
•         FileInputFormat.setInputPaths(conf, new Path("/home/hadoop/DataSet/Hadoop/WordCount"));
  
•         FileOutputFormat.setOutputPath(conf, new Path(outputPath));
  
•         JobClient.runJob(conf);         //运行一个job
  
•     }
  
• 
  
• }
  

　　上面是独立的一个Job，完成第一步。为了能紧接着完成第二步，我们需要在原来的基础上进行修改。
　　为了方便理解，上面的输入的例子如下：
　　Java代码

• accessed    3
  
• accessible  4
  
• accomplish  1
  
• accounting  7
  
• accurately  1
  
• acquire 1
  
• across  1
  
• actual  1
  
• actually    1
  
• add 3
  
• added   2
  
• addition    1
  
• additional  4
  

　　old api 的实现方式并不支持 setup() / cleanup() 操作这一点非常不好，因此在有可能的情况下最好还是要迁移到Hadoop 2.X
　　新的API会方便简洁很多
　　下面是增加了一个Mapper 来过滤
　　Java代码

• public static class RangeFilterMapper extends MapReduceBase implements Mapper<Text, IntWritable, Text, IntWritable> {
  
• 
  
•     @Override
  
•     public void map(Text key, IntWritable value,
  
•             OutputCollector<Text, IntWritable> output, Reporter reporter)
  
•             throws IOException {
  
• 
  
•         if(value.get() >= LOW_LIMIT) {
  
•             output.collect(key, value);
  
•         }
  
• 
  
•     }
  
• }
  

　　这个Mapper做的事情很简单，就是针对每个key，如果他的value > LOW_LIMIT 那么就输出
　　所以，目前为止，任务链如下：
　　TokenizerMapper -> TokenizeReducer -> RangeFilterMapper
　　所以我们的main函数改成下面的样子：
　　Java代码

• public static void main(String[] args) throws IOException {
  
• 
  
• 
  
•     JobConf conf = new JobConf(ChainedJobs.class);
  
•     conf.setJobName("wordcount");           //设置一个用户定义的job名称
  
• //        conf.setOutputKeyClass(Text.class);    //为job的输出数据设置Key类
  
• //        conf.setOutputValueClass(IntWritable.class);   //为job输出设置value类
  
• //        conf.setMapperClass(TokenizeMapper.class);         //为job设置Mapper类
  
• //        conf.setCombinerClass(TokenizeReducer.class);      //为job设置Combiner类
  
• //        conf.setReducerClass(TokenizeReducer.class);        //为job设置Reduce类
  
• //        conf.setInputFormat(TextInputFormat.class);    //为map-reduce任务设置InputFormat实现类
  
• //        conf.setOutputFormat(TextOutputFormat.class);  //为map-reduce任务设置OutputFormat实现类
  
• 
  
•     // Step1 : mapper forr word count
  
•     JobConf wordCountMapper  = new JobConf(false);
  
•     ChainMapper.addMapper(conf,
  
•             TokenizeMapper.class,
  
•             LongWritable.class,     // input key type
  
•             Text.class,             // input value type
  
•             Text.class,             // output key type
  
•             IntWritable.class,      // output value type
  
•             false,                  //byValue or byRefference 传值还是传引用
  
•             wordCountMapper);
  
• 
  
•     // Step2: reducer for word count
  
•     JobConf wordCountReducer  = new JobConf(false);
  
•     ChainReducer.setReducer(conf,
  
•             TokenizeReducer.class,
  
•             Text.class,
  
•             IntWritable.class,
  
•             Text.class,
  
•             IntWritable.class,
  
•             false,
  
•             wordCountReducer);
  
• 
  
•         // Step3: mapper used as filter
  
•     JobConf rangeFilterMapper  = new JobConf(false);
  
•     ChainReducer.addMapper(conf,
  
•             RangeFilterMapper.class,
  
•             Text.class,
  
•             IntWritable.class,
  
•             Text.class,
  
•             IntWritable.class,
  
•             false,
  
•             rangeFilterMapper);
  
• 
  
• 
  
•     // Remove output folder before run job(s)
  
•     FileSystem fs=FileSystem.get(conf);
  
•     String outputPath = "/home/hadoop/DataSet/Hadoop/WordCount-OUTPUT";
  
•     Path op=new Path(outputPath);
  
•     if (fs.exists(op)) {
  
•         fs.delete(op, true);
  
•         System.out.println("存在此输出路径，已删除！！！");
  
•     }
  
• 
  
•     FileInputFormat.setInputPaths(conf, new Path("/home/hadoop/DataSet/Hadoop/WordCount"));
  
•     FileOutputFormat.setOutputPath(conf, new Path(outputPath));
  
•     JobClient.runJob(conf);         //运行一个job
  
• }
  

　　下面是运行结果的一部分：
　　Java代码

• a   40
  
• and 26
  
• are 12
  
• as  6
  
• be  7
  
• been    8
  
• but 5
  
• by  5
  
• can 12
  
• change  5
  
• data    5
  
• files   7
  
• for 28
  
• from    5
  
• has 7
  
• have    8
  
• if  6
  
• in  27
  
• is  16
  
• it  13
  
• more    8
  
• not 5
  
• of  23
  
• on  5
  
• outputs 5
  
• see 6
  
• so  11
  
• that    11
  
• the 54
  

　　可以看到，英文之中，如果NLP不去除停用词（a, the, for ...） 等，效果确实会被大大的影响。