MapReduce的分區與 分組二次排序

原創做品，容許轉載，轉載時請務必以超連接形式標明文章  原始出處 、做者信息和本聲明。不然將追究法律責任。 http://computerdragon.blog.51cto.com/6235984/1287721

問題描述：

輸入文件格式以下：

name1    2

name3    4

name1    6

name1    1

name3    3

name1    0

要求輸出的文件格式以下：

name1    0，1，2，6

name3    3，4

要求是按照第一列分組，name1與name3也是按照順序排列的，組內升序排序。

思路：

常規的輸出，沒法排序key所對應的多個值的順序。爲了排序組內中的值，須要將key與value放在同一個組。Job中有兩個方法setGroupingComparatorClass和setSortComparatorClass，能夠利用這兩個方法來實現組內排序。可是這些排序都是基於key的，則就要將key和value定義成組合鍵。

可是必需要保證第一列相同的所有都放在同一個分區中，則就須要自定義分區，分區的時候只考慮第一列的值。因爲partitioner僅僅能保證每個reducer接受同一個name的全部記錄，可是reducer仍然是經過鍵進行分組的分區，也就說該分區中仍是按照鍵來分紅不一樣的組，還須要分組只參考name值

先按照name分組，再在name中內部進行排序。

解決方法：

運用自定義組合鍵的策略，將name和1定義爲一個組合鍵。在分區的時候只參考name的值，即繼承partitioner。

 因爲要按照name分組，則就須要定義分組策略，而後設置setGroupingComparatorClass。

setGroupingComparatorClass主要定義哪些key能夠放置在一組，分組的時候會對組合鍵進行比較，因爲這裏只須要考慮組合鍵中的一個值，則定義實現一個WritableComparator，設置比較策略。

對於組內的排序，能夠利用setSortComparatorClass來實現，

這個方法主要用於定義key如何進行排序在它們傳遞給reducer以前，

這裏就能夠來進行組內排序。

具體代碼：

     Hadoop版本號：hadoop1.1.2

自定義組合鍵

 1 package whut;
 2 import java.io.DataInput;
 3 import java.io.DataOutput;
 4 import java.io.IOException;
 5 import org.apache.hadoop.io.IntWritable;
 6 import org.apache.hadoop.io.Text;
 7 import org.apache.hadoop.io.WritableComparable;
 8 //自定義組合鍵策略
 9 //java基本類型數據
10 public class TextInt implements WritableComparable{
11     //直接利用java的基本數據類型
12     private String firstKey;
13     private int secondKey;
14     //必需要有一個默認的構造函數
15     public String getFirstKey() {
16         return firstKey;
17     }
18     public void setFirstKey(String firstKey) {
19         this.firstKey = firstKey;
20     }
21     public int getSecondKey() {
22         return secondKey;
23     }
24     public void setSecondKey(int secondKey) {
25         this.secondKey = secondKey;
26     }
27                                                                                                                                                                           
28     @Override
29     public void write(DataOutput out) throws IOException {
30         // TODO Auto-generated method stub
31         out.writeUTF(firstKey);
32         out.writeInt(secondKey);
33     }
34     @Override
35     public void readFields(DataInput in) throws IOException {
36         // TODO Auto-generated method stub
37         firstKey=in.readUTF();
38         secondKey=in.readInt();
39     }
40     //map的鍵的比較就是根據這個方法來進行的
41     @Override
42     public int compareTo(Object o) {
43         // TODO Auto-generated method stub
44         TextInt ti=(TextInt)o;
45         //利用這個來控制升序或降序
46         //this本對象寫在前面表明是升序
47         //this本對象寫在後面表明是降序
48         return this.getFirstKey().compareTo(ti.getFirstKey());
49     }
50 }

View Code

分組策略

 1 package whut;
 2 import org.apache.hadoop.io.WritableComparable;
 3 import org.apache.hadoop.io.WritableComparator;
 4 //主要就是對於分組進行排序，分組只按照組建鍵中的一個值進行分組
 5 public class TextComparator extends WritableComparator {
 6     //必需要調用父類的構造器
 7     protected TextComparator() {
 8         super(TextInt.class,true);//註冊comparator
 9     }
10     @Override
11     public int compare(WritableComparable a, WritableComparable b) {
12         // TODO Auto-generated method stub
13         TextInt ti1=(TextInt)a;
14         TextInt ti2=(TextInt)b;
15         return ti1.getFirstKey().compareTo(ti2.getFirstKey());
16     }
17 }

View Code

組內排序策略

 1 package whut;
 2 import org.apache.hadoop.io.WritableComparable;
 3 import org.apache.hadoop.io.WritableComparator;
 4 //分組內部進行排序，按照第二個字段進行排序
 5 public class TextIntComparator extends WritableComparator {
 6     public TextIntComparator()
 7     {
 8         super(TextInt.class,true);
 9     }
10     //這裏能夠進行排序的方式管理
11     //必須保證是同一個分組的
12     //a與b進行比較
13     //若是a在前b在後，則會產生升序
14     //若是a在後b在前，則會產生降序
15     @Override
16     public int compare(WritableComparable a, WritableComparable b) {
17         // TODO Auto-generated method stub
18         TextInt ti1=(TextInt)a;
19         TextInt ti2=(TextInt)b;
20         //首先要保證是同一個組內，同一個組的標識就是第一個字段相同
21         if(!ti1.getFirstKey().equals(ti2.getFirstKey()))
22            return ti1.getFirstKey().compareTo(ti2.getFirstKey());
23         else
24            return ti2.getSecondKey()-ti1.getSecondKey();//0,-1,1
25     }
26                                                                                                                                                           
27 }

View Code

分區策略

 1 package whut;
 2 import org.apache.hadoop.io.IntWritable;
 3 import org.apache.hadoop.mapreduce.Partitioner;
 4 //參數爲map的輸出類型
 5 public class KeyPartitioner extends Partitioner<TextInt, IntWritable> {
 6     @Override
 7     public int getPartition(TextInt key, IntWritable value, int numPartitions) {
 8         // TODO Auto-generated method stub
 9         return (key.getFirstKey().hashCode()&Integer.MAX_VALUE)%numPartitions;
10     }
11 }

View Code

MapReduce策略

  1 package whut;
  2 import java.io.IOException;
  3 import org.apache.hadoop.conf.Configuration;
  4 import org.apache.hadoop.conf.Configured;
  5 import org.apache.hadoop.fs.Path;
  6 import org.apache.hadoop.io.IntWritable;
  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.Mapper.Context;
 12 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
 13 import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
 14 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 15 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
 16 import org.apache.hadoop.util.Tool;
 17 import org.apache.hadoop.util.ToolRunner;
 18 //須要對數據進行分組以及組內排序的時候
 19 public class SortMain extends Configured implements Tool{
 20     //這裏設置輸入文格式爲KeyValueTextInputFormat
 21     //name1 5
 22     //默認輸入格式都是Text，Text
 23     public static class GroupMapper extends
 24        Mapper<Text, Text, TextInt, IntWritable>  {
 25         public IntWritable second=new IntWritable();
 26         public TextInt tx=new TextInt();
 27         @Override
 28         protected void map(Text key, Text value, Context context)
 29                 throws IOException, InterruptedException {
 30             String lineKey=key.toString();
 31             String lineValue=value.toString();
 32             int lineInt=Integer.parseInt(lineValue);
 33             tx.setFirstKey(lineKey);
 34             tx.setSecondKey(lineInt);
 35             second.set(lineInt);
 36             context.write(tx, second);
 37         }
 38     }
 39     //設置reduce
 40     public static class GroupReduce extends Reducer<TextInt, IntWritable, Text, Text>
 41     {
 42         @Override
 43         protected void reduce(TextInt key, Iterable<IntWritable> values,
 44                Context context)
 45                 throws IOException, InterruptedException {
 46             StringBuffer sb=new StringBuffer();
 47             for(IntWritable val:values)
 48             {
 49                 sb.append(val+",");
 50             }
 51             if(sb.length()>0)
 52             {
 53                 sb.deleteCharAt(sb.length()-1);
 54             }
 55             context.write(new Text(key.getFirstKey()), new Text(sb.toString()));
 56         }
 57     }
 58                                                                                                                                         
 59     @Override
 60     public int run(String[] args) throws Exception {
 61         // TODO Auto-generated method stub
 62         Configuration conf=getConf();
 63         Job job=new Job(conf,"SecondarySort");
 64         job.setJarByClass(SortMain.class);
 65         // 設置輸入文件的路徑，已經上傳在HDFS
 66         FileInputFormat.addInputPath(job, new Path(args[0]));
 67         // 設置輸出文件的路徑，輸出文件也存在HDFS中，可是輸出目錄不能已經存在
 68         FileOutputFormat.setOutputPath(job, new Path(args[1]));
 69                                                                                                                                             
 70         job.setMapperClass(GroupMapper.class);
 71         job.setReducerClass(GroupReduce.class);
 72         //設置分區方法
 73         job.setPartitionerClass(KeyPartitioner.class);
 74                                                                                                                                             
 75         //下面這兩個都是針對map端的
 76         //設置分組的策略，哪些key能夠放置到一組中
 77         job.setGroupingComparatorClass(TextComparator.class);
 78         //設置key如何進行排序在傳遞給reducer以前.
 79         //這裏就能夠設置對組內如何排序的方法
 80         /*************關鍵點**********/
 81         job.setSortComparatorClass(TextIntComparator.class);
 82         //設置輸入文件格式
 83         job.setInputFormatClass(KeyValueTextInputFormat.class);
 84         //使用默認的輸出格式即TextInputFormat
 85         //設置map的輸出key和value類型
 86         job.setMapOutputKeyClass(TextInt.class);
 87         job.setMapOutputValueClass(IntWritable.class);
 88         //設置reduce的輸出key和value類型
 89         //job.setOutputFormatClass(TextOutputFormat.class);
 90         job.setOutputKeyClass(Text.class);
 91         job.setOutputValueClass(Text.class);
 92         job.waitForCompletion(true);
 93         int exitCode=job.isSuccessful()?0:1;
 94         return exitCode;
 95     }
 96                                                                                                                                         
 97     public static void main(String[] args)  throws Exception
 98     {
 99        int exitCode=ToolRunner.run(new SortMain(), args);
100        System.exit(exitCode);
101     }
102 }

View Code

注意事項

   1，設置分組排序按照升序仍是降序是在自定義WritableComparable中的compareTo()方法實現的，具體升序或者降序的設置在代碼中已經註釋說明

   2，設置組內值進行升序仍是降序的排序是在組內排序策略中的compare()方法註釋說明的。

   3，這裏同時最重要的一點是，將第二列即放在組合鍵中，又做爲value，這樣對於組合鍵排序也就至關於對於value進行排序了。

   4，在自定義組合鍵的時候，對於組合鍵中的數據的基本類型能夠採用Java的基本類型也能夠採用Hadoop的基本數據類型，對於Hadoop的基本數據類型必定要記得初始化new一個基本數據類型對象。對於組合鍵類，必需要有默認的構造方法。

本文出自 「在雲端的追夢」 博客，請務必保留此出處http://computerdragon.blog.51cto.com/6235984/1287721