Mapreduce实例——ChainMapReduce

原理

一些复杂的任务难以用一次MapReduce处理完成，需要多次MapReduce才能完成任务。Hadoop2.0开始MapReduce作业支持链式处理，类似于工厂的的生产线，每一个阶段都有特定的任务要处理，比如提供原配件——>组装——打印出厂日期，等等。通过这样进一步的分工，从而提高了生成效率，我们Hadoop中的链式MapReduce也是如此，这些Mapper可以像水流一样，一级一级向后处理，有点类似于Linux的管道。前一个Mapper的输出结果直接可以作为下一个Mapper的输入，形成一个流水线。

链式MapReduce的执行规则：整个Job中只能有一个Reducer，在Reducer前面可以有一个或者多个Mapper，在Reducer的后面可以有0个或者多个Mapper。

Hadoop2.0支持的链式处理MapReduce作业有一下三种：

（1）顺序链接MapReduce作业

类似于Unix中的管道：mapreduce-1 | mapreduce-2 | mapreduce-3 ......，每一个阶段创建一个job，并将当前输入路径设为前一个的输出。在最后阶段删除链上生成的中间数据。

（2）具有复杂依赖的MapReduce链接

若mapreduce-1处理一个数据集， mapreduce-2 处理另一个数据集，而mapreduce-3对前两个做内部连结。这种情况通过Job和JobControl类管理非线性作业间的依赖。如x.addDependingJob(y)意味着x在y完成前不会启动。

（3）预处理和后处理的链接

一般将预处理和后处理写为Mapper任务。可以自己进行链接或使用ChainMapper和ChainReducer类，生成得作业表达式类似于：

MAP+ | REDUCE | MAP*

如以下作业： Map1 | Map2 | Reduce | Map3 | Map4，把Map2和Reduce视为MapReduce作业核心。Map1作为前处理，Map3， Map4作为后处理。ChainMapper使用模式：（预处理作业），ChainReducer使用模式：（设置Reducer并添加后处理Mapper）

本实验中用到的就是第三种作业模式：预处理和后处理的链接，生成得作业表达式类似于 Map1 | Map2 | Reduce | Map3

环境

Linux Ubuntu 14.04

jdk-7u75-linux-x64

hadoop-2.6.0-cdh5.4.5

hadoop-2.6.0-eclipse-cdh5.4.5.jar

eclipse-java-juno-SR2-linux-gtk-x86_64

内容

练习使用ChainMapReduce处理文件，现有某电商一天商品浏览情况数据goods_0，功能为在第一个Mapper里面过滤掉点击量大于600的商品，在第二个Mapper中过滤掉点击量在100~600之间的商品，Reducer里面进行分类汇总并输出，在Reducer后的Mapper里过滤掉商品名长度大于或等于3的商品

实验数据如下：

表goods_0，包含两个字段（商品名称，点击量），分隔符为" "。

  1. 商品名称  点击量  
  2. 袜子       189  
  3. 毛衣       600  
  4. 裤子       780  
  5. 鞋子       30  
  6. 呢子外套    90  
  7. 牛仔外套   130  
  8. 羽绒服    7  
  9. 帽子      21  
  10. 帽子      6  
  11. 羽绒服   12  

结果数据如下：

  1. 商品名称  点击量  
  2. 帽子       27.0  
  3. 鞋子       30.0  

实验步骤

1，切换到/apps/hadoop/sbin目录下，开启Hadoop。

  1. cd /apps/hadoop/sbin  
  2. ./start-all.sh  

2，在Linux本地新建/data/mapreduce10目录。

  1. mkdir -p /data/mapreduce10  

3，在Linux中切换到/data/mapreduce10目录下，用wget命令从http://192.168.1.100:60000/allfiles/mapreduce10/goods_0网址上下载文本文件goods_0。

  1. cd /data/mapreduce10  
  2. wget http://192.168.1.100:60000/allfiles/mapreduce10/goods_0  

然后在当前目录下用wget命令从http://192.168.1.100:60000/allfiles/mapreduce10/hadoop2lib.tar.gz网址上下载项目用到的依赖包。

  1. wget http://192.168.1.100:60000/allfiles/mapreduce10/hadoop2lib.tar.gz  

将hadoop2lib.tar.gz解压到当前目录下。

  1. tar zxvf hadoop2lib.tar.gz  

4，首先在HDFS上新建/mymapreduce10/in目录，然后将Linux本地/data/mapreduce10目录下的goods_0文件导入到HDFS的/mymapreduce10/in目录中。

  1. hadoop fs -mkdir -p /mymapreduce10/in  
  2. hadoop fs -put /data/mapreduce10/goods_0 /mymapreduce10/in  

5，新建Java Project项目，项目名为mapreduce10。

在mapreduce10项目下新建mapreduce包。

在mapreduce包下新建ChainMapReduce类。

6，添加项目所需依赖的jar包

右键项目，新建一个文件夹，用于存放项目所需的jar包。

将/data/mapreduce10目录下，hadoop2lib目录中的jar包，拷贝到eclipse中mapreduce10项目的hadoop2lib目录下。

选中所有项目hadoop2lib目录下所有jar包，单击右键选择Build Path=>Add to Build Path。

7，编写程序代码，并描述其设计思路。

mapreduce执行的大体流程如下图所示：

由上图可知，ChainMapReduce的执行流程为：①首先将文本文件中的数据通过InputFormat实例切割成多个小数据集InputSplit，然后通过RecordReader实例将小数据集InputSplit解析为<key,value>的键值对并提交给Mapper1；②Mapper1里的map函数将输入的value进行切割，把商品名字段作为key值，点击数量字段作为value值，筛选出value值小于等于600的<key,value>，将<key,value>输出给Mapper2，③Mapper2里的map函数再筛选出value值小于100的<key,value>，并将<key,value>输出；④Mapper2输出的<key,value>键值对先经过shuffle，将key值相同的所有value放到一个集合，形成<key,value-list>，然后将所有的<key,value-list>输入给Reducer；⑤Reducer里的reduce函数将value-list集合中的元素进行累加求和作为新的value，并将<key,value>输出给Mapper3，⑥Mapper3里的map函数筛选出key值小于3个字符的<key,value>，并将<key,value>以文本的格式输出到hdfs上。该ChainMapReduce的Java代码主要分为四个部分，分别为：FilterMapper1，FilterMapper2，SumReducer，FilterMapper3。

FilterMapper1代码

  1. public static class FilterMapper1 extends Mapper<LongWritable, Text, Text, DoubleWritable> {  
  2.         private Text outKey = new Text();    //声明对象outKey  
  3.         private DoubleWritable outValue = new DoubleWritable();    //声明对象outValue  
  4.         @Override  
  5.         protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, DoubleWritable>.Context context)  
  6.         throws IOException,InterruptedException {  
  7.             String line = value.toString();  
  8.             if (line.length() > 0) {  
  9.                 String[] splits = line.split(" ");  //按行对内容进行切分  
  10.                 double visit = Double.parseDouble(splits[1].trim());  
  11.                 if (visit <= 600) {    //if循环，判断visit是否小于等于600  
  12.                     outKey.set(splits[0]);  
  13.                     outValue.set(visit);  
  14.                     context.write(outKey, outValue);  //调用context的write方法  
  15.                 }  
  16.             }  
  17.         }  
  18.     }  

首先定义输出的key和value的类型，然后在map方法中获取文本行内容，用Split(" ")对行内容进行切分，把包含点击量的字段转换成double类型并赋值给visit，用if判断，如果visit小于等于600，则设置商品名称字段作为key,设置该visit作为value，用context的write方法输出<key,value>。

FilterMapper2代码

  1. public static class FilterMapper2 extends Mapper<Text, DoubleWritable, Text, DoubleWritable> {  
  2.        @Override  
  3.        protected void map(Text key, DoubleWritable value, Mapper<Text, DoubleWritable, Text, DoubleWritable>.Context context)  
  4.     throws IOException,InterruptedException {  
  5.            if (value.get() < 100) {  
  6.                context.write(key, value);  
  7.            }  
  8.        }  
  9.    }  

接收mapper1传来的数据，通过value.get()获取输入的value值，再用if判断如果输入的value值小于100，则直接将输入的key赋值给输出的key，输入的value赋值给输出的value，输出<key,value>。

SumReducer代码

  1. public  static class SumReducer extends Reducer<Text, DoubleWritable, Text, DoubleWritable> {  
  2.        private DoubleWritable outValue = new DoubleWritable();  
  3.        @Override  
  4.        protected void reduce(Text key, Iterable<DoubleWritable> values, Reducer<Text, DoubleWritable, Text, DoubleWritable>.Context context)  
  5.    throws IOException, InterruptedException {  
  6.    double sum = 0;  
  7.    for (DoubleWritable val : values) {  
  8.    sum += val.get();  
  9.    }  
  10.    outValue.set(sum);  
  11.    context.write(key, outValue);  
  12.    }  
  13.    }  

FilterMapper2输出的<key,value>键值对先经过shuffle，将key值相同的所有value放到一个集合，形成<key,value-list>，然后将所有的<key,value-list>输入给SumReducer。在reduce函数中，用增强版for循环遍历value-list中元素，将其数值进行累加并赋值给sum，然后用outValue.set(sum)方法把sum的类型转变为DoubleWritable类型并将sum设置为输出的value，将输入的key赋值给输出的key，最后用context的write()方法输出<key,value>。

FilterMapper3代码

  1. public  static class FilterMapper3 extends Mapper<Text, DoubleWritable, Text, DoubleWritable> {  
  2.        @Override  
  3.        protected void map(Text key, DoubleWritable value, Mapper<Text, DoubleWritable, Text, DoubleWritable>.Context context)  
  4.     throws IOException, InterruptedException {  
  5.            if (key.toString().length() < 3) {  //for循环，判断key值是否大于3  
  6.                System.out.println("写出去的内容为：" + key.toString() +"++++"+ value.toString());  
  7.                context.write(key, value);  
  8.            }  
  9.        }  
  10.    }  

接收reduce传来的数据，通过key.toString().length()获取key值的字符长度，再用if判断如果key值的字符长度小于3，则直接将输入的key赋值给输出的key，输入的value赋值给输出的value，输出<key，value>。

完整代码

  1. package mapreduce;  
  2. import java.io.IOException;  
  3. import java.net.URI;  
  4. import org.apache.hadoop.conf.Configuration;  
  5. import org.apache.hadoop.fs.Path;  
  6. import org.apache.hadoop.io.LongWritable;  
  7. import org.apache.hadoop.io.Text;  
  8. import org.apache.hadoop.mapreduce.Job;  
  9. import org.apache.hadoop.mapreduce.Mapper;  
  10. import org.apache.hadoop.mapreduce.Reducer;  
  11. import org.apache.hadoop.mapreduce.lib.chain.ChainMapper;  
  12. import org.apache.hadoop.mapreduce.lib.chain.ChainReducer;  
  13. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
  14. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;  
  15. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
  16. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;  
  17. import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;  
  18. import org.apache.hadoop.fs.FileSystem;  
  19. import org.apache.hadoop.io.DoubleWritable;  
  20. public class ChainMapReduce {  
  21.     private static final String INPUTPATH = "hdfs://localhost:9000/mymapreduce10/in/goods_0";  
  22.     private static final String OUTPUTPATH = "hdfs://localhost:9000/mymapreduce10/out";  
  23.     public static void main(String[] args) {  
  24.         try {  
  25.             Configuration conf = new Configuration();  
  26.             FileSystem fileSystem = FileSystem.get(new URI(OUTPUTPATH), conf);  
  27.             if (fileSystem.exists(new Path(OUTPUTPATH))) {  
  28.                 fileSystem.delete(new Path(OUTPUTPATH), true);  
  29.             }  
  30.             Job job = new Job(conf, ChainMapReduce.class.getSimpleName());  
  31.             FileInputFormat.addInputPath(job, new Path(INPUTPATH));  
  32.             job.setInputFormatClass(TextInputFormat.class);  
  33.             ChainMapper.addMapper(job, FilterMapper1.class, LongWritable.class, Text.class, Text.class, DoubleWritable.class, conf);  
  34.             ChainMapper.addMapper(job, FilterMapper2.class, Text.class, DoubleWritable.class, Text.class, DoubleWritable.class, conf);  
  35.             ChainReducer.setReducer(job, SumReducer.class, Text.class, DoubleWritable.class, Text.class, DoubleWritable.class, conf);  
  36.             ChainReducer.addMapper(job, FilterMapper3.class, Text.class, DoubleWritable.class, Text.class, DoubleWritable.class, conf);  
  37.             job.setMapOutputKeyClass(Text.class);  
  38.             job.setMapOutputValueClass(DoubleWritable.class);  
  39.             job.setPartitionerClass(HashPartitioner.class);  
  40.             job.setNumReduceTasks(1);  
  41.             job.setOutputKeyClass(Text.class);  
  42.             job.setOutputValueClass(DoubleWritable.class);  
  43.             FileOutputFormat.setOutputPath(job, new Path(OUTPUTPATH));  
  44.             job.setOutputFormatClass(TextOutputFormat.class);  
  45.             System.exit(job.waitForCompletion(true) ? 0 : 1);  
  46.         } catch (Exception e) {  
  47.             e.printStackTrace();  
  48.         }  
  49.     }  
  50.     public static class FilterMapper1 extends Mapper<LongWritable, Text, Text, DoubleWritable> {  
  51.         private Text outKey = new Text();  
  52.         private DoubleWritable outValue = new DoubleWritable();  
  53.         @Override  
  54.         protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, DoubleWritable>.Context context)  
  55.         throws IOException,InterruptedException {  
  56.             String line = value.toString();  
  57.             if (line.length() > 0) {  
  58.                 String[] splits = line.split(" ");  
  59.                 double visit = Double.parseDouble(splits[1].trim());  
  60.                 if (visit <= 600) {  
  61.                     outKey.set(splits[0]);  
  62.                     outValue.set(visit);  
  63.                     context.write(outKey, outValue);  
  64.                 }  
  65.             }  
  66.         }  
  67.     }  
  68.     public static class FilterMapper2 extends Mapper<Text, DoubleWritable, Text, DoubleWritable> {  
  69.         @Override  
  70.         protected void map(Text key, DoubleWritable value, Mapper<Text, DoubleWritable, Text, DoubleWritable>.Context context)  
  71.         throws IOException,InterruptedException {  
  72.             if (value.get() < 100) {  
  73.                 context.write(key, value);  
  74.             }  
  75.         }  
  76.     }  
  77.     public  static class SumReducer extends Reducer<Text, DoubleWritable, Text, DoubleWritable> {  
  78.         private DoubleWritable outValue = new DoubleWritable();  
  79.         @Override  
  80.         protected void reduce(Text key, Iterable<DoubleWritable> values, Reducer<Text, DoubleWritable, Text, DoubleWritable>.Context context)  
  81.     throws IOException, InterruptedException {  
  82.     double sum = 0;  
  83.     for (DoubleWritable val : values) {  
  84.     sum += val.get();  
  85.     }  
  86.     outValue.set(sum);  
  87.     context.write(key, outValue);  
  88.     }  
  89.     }  
  90.     public  static class FilterMapper3 extends Mapper<Text, DoubleWritable, Text, DoubleWritable> {  
  91.     @Override  
  92.     protected void map(Text key, DoubleWritable value, Mapper<Text, DoubleWritable, Text, DoubleWritable>.Context context)  
  93.     throws IOException, InterruptedException {  
  94.     if (key.toString().length() < 3) {  
  95.     System.out.println("写出去的内容为：" + key.toString() +"++++"+ value.toString());  
  96.     context.write(key, value);  
  97.     }  
  98.     }  
  99.     
  100.     }  
  101.     
  102.     }  

8，在ChainMapReduce类文件中，右键并点击=>Run As=>Run on Hadoop选项，将MapReduce任务提交到Hadoop中。

9，待执行完毕后，进入命令模式下，在hdfs上/mymapreduce10/out中查看实验结果。

  1. hadoop fs -ls /mymapreduce10/out  
  2. hadoop fs -cat /mymapreduce10/out/part-r-00000  

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/aishanyishi/p/10304868.html