Mapreduce实例——排序
原理
Map、Reduce任务中Shuffle和排序的过程图如下：
流程分析：
1.Map端：
（1）每个输⼊分⽚会让⼀个map任务来处理，默认情况下，以HDFS的⼀个块的⼤⼩（默认为64M）为⼀个分⽚，当然我们也可以设置块的⼤⼩。map输出的结果会暂且放在⼀个环形内存缓冲区中（该缓冲区的⼤⼩默认为100M，由io.sort.mb属性控制），当该缓冲区快要溢出时（默认为缓冲区⼤⼩的80%，由io.sort.spill.percent属性控制），会在本地⽂件系统中创建⼀个溢出⽂件，将该缓冲区中的数据写⼊这个⽂件。
（2）在写⼊磁盘之前，线程⾸先根据reduce任务的数⽬将数据划分为相同数⽬的分区，也就是⼀个reduce任务对应⼀个分区的数据。这样做是为了避免有些reduce任务分配到⼤量数据，⽽有些reduce任务却分到很少数据，甚⾄没有分到数据的尴尬局⾯。其实分区就是对数据进⾏hash的过程。然后对每个分区中的数据进⾏排序，如果此时设置了Combiner，将排序后的结果进⾏Combia操作，这样做的⽬的
是让尽可能少的数据写⼊到磁盘。
（3）当map任务输出最后⼀个记录时，可能会有很多的溢出⽂件，这时需要将这些⽂件合并。合并的过程中会不断地进⾏排序和combia操作，⽬的有两个：①尽量减少每次写⼊磁盘的数据量。②尽量减少下⼀复制阶段⽹络传输的数据量。最后合并成了⼀个已分区且已排序的⽂件。为了减少⽹络传输的数据量，这⾥可以将数据压缩，只要将mapredpress.map.out设置为true就可以了。
（4）将分区中的数据拷贝给相对应的reduce任务。有⼈可能会问：分区中的数据怎么知道它对应的reduce是哪个呢？其实map任务⼀直和其⽗TaskTracker保持联系，⽽TaskTracker⼜⼀直和JobTracker保持⼼跳。所以JobTracker中保存了整个集中的宏观信息。只要reduce任务向JobTracker获取对应的map输出位置就ok了哦。
到这⾥，map端就分析完了。那到底什么是Shuffle呢？Shuffle的中⽂意思是"洗牌"，如果我们这样看：⼀个map产⽣的数据，结果通过hash 过程分区却分配给了不同的reduce任务，是不是⼀个对数据洗牌的过程呢？
2.Reduce端：
（1）Reduce会接收到不同map任务传来的数据，并且每个map传来的数据都是有序的。如果reduce
端接受的数据量相当⼩，则直接存储在内存中（缓冲区⼤⼩由mapred.job.shuffle.input.buffer.percent属性控制，表⽰⽤作此⽤途的堆空间的百分⽐），如果数据量超过了该缓冲区⼤⼩的⼀定⽐例（由mapred.percent决定），则对数据合并后溢写到磁盘中。
（2）随着溢写⽂件的增多，后台线程会将它们合并成⼀个更⼤的有序的⽂件，这样做是为了给后⾯的合并节省时间。其实不管在map端还是reduce端，MapReduce都是反复地执⾏排序，合并操作，现在终于明⽩了有些⼈为什么会说：排序是hadoop的灵魂。
（3）合并的过程中会产⽣许多的中间⽂件（写⼊磁盘了），但MapReduce会让写⼊磁盘的数据尽可能地少，并且最后⼀次合并的结果并没有写⼊磁盘，⽽是直接输⼊到reduce函数。
熟悉MapReduce的⼈都知道：排序是MapReduce的天然特性！在数据达到reducer之前，MapReduce框架已经对这些数据按键排序了。但是在使⽤之前，⾸先需要了解它的默认排序规则。它是按照key值进⾏排序的，如果key为封装的int为IntWritable类型，那么MapReduce按照数字⼤⼩对key排序，如果Key为封装String的Text类型，那么MapReduce将按照数据字典顺序对字符排序。
了解了这个细节，我们就知道应该使⽤封装int的Intwritable型数据结构了，也就是在map这⾥，将读⼊的数据中要排序的字段转化为Intwritable型，然后作为key值输出（不排序的字段作为value）。reduce阶段拿到<key，value-list>之后，将输⼊的key作为的输出key，并根据value-list中的元素的个数决定
输出的次数。
实验环境
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
实验内容
在电商⽹站上，当我们进⼊某电商页⾯⾥浏览商品时，就会产⽣⽤户对商品访问情况的数据，名为goods_visit1，goods_visit1中包含（商品id ，点击次数）两个字段，内容以"\t"分割，由于数据量很⼤，所以为了⽅便统计我们只截取它的⼀部分数据，内容如下：
1. 商品id  点击次数
2. 1010037 100
3. 1010102 100
4. 1010152 97
5. 1010178 96
6. 1010280 104
7. 1010320 103
8. 1010510 104
9. 1010603 96
10. 1010637 97
要求我们编写mapreduce程序来对商品点击次数有低到⾼进⾏排序。
实验结果数据如下：
1. 点击次数商品ID
2. 96  1010603
3. 96  1010178
4. 97  1010637
5. 97  1010152
6. 100 1010102
7. 100 1010037
8. 103 1010320
9. 104 1010510
10. 104 1010280
实验步骤
1.切换到/apps/hadoop/sbin⽬录下，开启Hadoop。
1. cd /apps/hadoop/sbin
2. ./start-all.sh
2.在Linux本地新建/data/mapreduce3⽬录。
1. mkdir -p /data/mapreduce3
1. cd /data/mapreduce3
2. wget 192.168.1.100:60000/allfiles/mapreduce3/goods_visit1
1. wget 19
2.168.1.100:60000/allfiles/mapreduce3/
将解压到当前⽬录下。
1. tar zxvf
4.⾸先在HDFS上新建/mymapreduce3/in⽬录，然后将Linux本地/data/mapreduce3⽬录下的goods_visit1⽂件导⼊到HDFS 的/mymapreduce3/in⽬录中。
1. hadoop fs -mkdir -p /mymapreduce3/in
2. hadoop fs -put /data/mapreduce3/goods_visit1 /mymapreduce3/in
5.新建Java Project项⽬，项⽬名为mapreduce3。
在mapreduce3项⽬下新建包，包名为mapreduce。
在mapreduce包下新建类，类名为OneSort。
6.添加项⽬所需依赖的jar包，右键单击项⽬新建⼀个⽂件夹，名为hadoop2lib，⽤于存放项⽬所需的jar包。
将/data/mapreduce3⽬录下hadoop2lib⽂件夹中的所有jar包，拷贝到eclipse中mapreduce3项⽬的hadoop2lib⽬录下。
选中hadoop2lib⽬录下所有jar包，单击右键，选择Build Path→Add to Build Path。
sort命令排序
7.编写Java代码，并描述其设计思路
在MapReduce过程中默认就有对数据的排序。它是按照key值进⾏排序的，如果key为封装int的IntWritable类型，那么MapReduce会按照数字⼤⼩对key排序，如果Key为封装String的Text类型，那么MapReduce将按照数据字典顺序对字符排序。在本例中我们⽤到第⼀种，key设置为IntWritable类型，其中MapReduce程序主要分为Map部分和Reduce部分。
Map部分代码
1. public static class Map extends Mapper<Object,Text,IntWritable,Text>{
2. private static Text goods=new Text();
3. private static IntWritable num=new IntWritable();
4. public void map(Object key,Text value,Context context) throws IOException, InterruptedException{
5.            String String();
6.            String arr[]=line.split("\t");
7.            num.set(Integer.parseInt(arr[1]));
8.            goods.set(arr[0]);
9.            context.write(num,goods);
10.        }
11.    }
在map端采⽤Hadoop默认的输⼊⽅式之后，将输⼊的value值⽤split()⽅法截取，把要排序的点击次数字段转化为IntWritable类型并设置为key，商品id字段设置为value，然后直接输出<key,value>。map输出的<key,value>先要经过shuffle过程把相同key值的所有value聚集起来形成<key,value-list>后交给reduce端。
Reduce部分代码
1. public static class Reduce extends Reducer<IntWritable,Text,IntWritable,Text>{
2. private static IntWritable result= new IntWritable();
3. //声明对象result
4. public void reduce(IntWritable key,Iterable<Text> values,Context context) throws IOException, InterruptedException{
5. for(Text val:values){
6.    context.write(key,val);
7.    }
8.    }
9.    }
reduce端接收到<key,value-list>之后，将输⼊的key直接复制给输出的key,⽤for循环遍历value-list并将⾥⾯的元素设置为输出的value，然后将<key,value>逐⼀输出，根据value-list中元素的个数决定输出的次数。
完整代码
1. package mapreduce;
2. import java.io.IOException;
3. import org.f.Configuration;
4. import org.apache.hadoop.fs.Path;
5. import org.apache.hadoop.io.IntWritable;
6. import org.apache.hadoop.io.Text;
7. import org.apache.hadoop.mapreduce.Job;
8. import org.apache.hadoop.mapreduce.Mapper;
9. import org.apache.hadoop.mapreduce.Reducer;
10. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
11. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
12. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
13. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
14. public class OneSort {
15. public static class Map extends Mapper<Object , Text , IntWritable,Text >{
16. private static Text goods=new Text();
17. private static IntWritable num=new IntWritable();
18. public void map(Object key,Text value,Context context) throws IOException, InterruptedException{
19.    String String();
20.    String arr[]=line.split("\t");
21.    num.set(Integer.parseInt(arr[1]));
22.    goods.set(arr[0]);
23.    context.write(num,goods);
24.    }
25.    }
26. public static class Reduce extends Reducer< IntWritable, Text, IntWritable, Text>{
27. private static IntWritable result= new IntWritable();
28. public void reduce(IntWritable key,Iterable<Text> values,Context context) throws IOException, InterruptedException{
29. for(Text val:values){
30.        context.write(key,val);
31.        }
32.        }
33.        }
34. public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException{
35.        Configuration conf=new Configuration();
36.        Job job =new Job(conf,"OneSort");
37.        job.setJarByClass(OneSort.class);
38.        job.setMapperClass(Map.class);
39.        job.setReducerClass(Reduce.class);
40.        job.setOutputKeyClass(IntWritable.class);
41.        job.setOutputValueClass(Text.class);
42.        job.setInputFormatClass(TextInputFormat.class);
43.        job.setOutputFormatClass(TextOutputFormat.class);
44.        Path in=new Path("hdfs://localhost:9000/mymapreduce3/in/goods_visit1");
45.        Path out=new Path("hdfs://localhost:9000/mymapreduce3/out");
46.        FileInputFormat.addInputPath(job,in);
47.        FileOutputFormat.setOutputPath(job,out);
48.        it(job.waitForCompletion(true) ? 0 : 1);
49.
50.        }
51.        }
8.在OneSort类⽂件中，右键并点击=>Run As=>Run on Hadoop选项，将MapReduce任务提交到Hadoop中。
9.待执⾏完毕后，进⼊命令模式下，在HDFS上/mymapreduce3/out中查看实验结果。
1. hadoop fs -ls /mymapreduce3/out
2. hadoop fs -cat /mymapreduce3/out/part-r-00000
窗体顶端
窗体底端

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。