码迷，mamicode.com
首页 > 其他好文 > 详细

MapReduce自定义分组Group

时间：2016-01-31 21:12:43      阅读：295      评论：0      收藏：0      [点我收藏+]

标签：

一：背景

在上一篇文章中我们可以对两列数据进行排序，即完成了当第一列相同时第二列数据升序排列的功能，现在我们需要进一步完善一个功能，那就是当第一列相同时求出第二列的最小值或最大值，Hadoop提供了自定义分组的功能，可以满足我们的需求。

 

二：技术实现

我们先来看看需求

#当第一列不相等时，第一列按升序排列，当第一列相等时，求出对应第二列的最小值

 

[java] view plain copy
 
  1. 3   3  
  2. 3   2  
  3. 3   1  
  4. 2   2  
  5. 2   1  
  6. 1   1  


输出结果应该是：

 

 

[java] view plain copy
 
  1. 1   1  
  2. 2   1  
  3. 3   1  



 

实现：

(1).自定义分组比较器继承RawComparator，实现compare()方法。

(2).在设置作业是设置job.setGroupingComparatorClass()。

 

代码如下：

 

[java] view plain copy
 
    1. public class MyGroupTest {  
    2.     // 定义输入路径  
    3.     private static final String INPUT_PATH = "hdfs://liaozhongmin:9000/data";  
    4.     // 定义输出路径  
    5.     private static final String OUT_PATH = "hdfs://liaozhongmin:9000/out";  
    6.   
    7.     public static void main(String[] args) {  
    8.   
    9.         try {  
    10.             // 创建配置信息  
    11.             Configuration conf = new Configuration();  
    12.   
    13.   
    14.             // 创建文件系统  
    15.             FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), conf);  
    16.             // 如果输出目录存在，我们就删除  
    17.             if (fileSystem.exists(new Path(OUT_PATH))) {  
    18.                 fileSystem.delete(new Path(OUT_PATH), true);  
    19.             }  
    20.   
    21.             // 创建任务  
    22.             Job job = new Job(conf, MyGroupTest.class.getName());  
    23.   
    24.             // 天龙八部1.1 设置输入目录和设置输入数据格式化的类  
    25.             FileInputFormat.setInputPaths(job, INPUT_PATH);  
    26.             job.setInputFormatClass(TextInputFormat.class);  
    27.   
    28.             //1.2 设置自定义Mapper类和设置map函数输出数据的key和value的类型  
    29.             job.setMapperClass(MyGroupMapper.class);  
    30.             job.setMapOutputKeyClass(CombineKey.class);  
    31.             job.setMapOutputValueClass(LongWritable.class);  
    32.               
    33.             //一定不要忘记设置自定义分组比较器的类(这一步是关键)  
    34.             job.setGroupingComparatorClass(MyGroupComparator.class);  
    35.               
    36.             //1.3 设置分区和reduce数量(reduce的数量，和分区的数量对应，因为分区为一个，所以reduce的数量也是一个)  
    37.             job.setPartitionerClass(HashPartitioner.class);  
    38.             job.setNumReduceTasks(1);  
    39.   
    40.             //1.4 排序、分组  
    41.             //1.5 归约  
    42.             //2.1 Shuffle把数据从Map端拷贝到Reduce端。  
    43.             //2.2 指定Reducer类和输出key和value的类型  
    44.             job.setReducerClass(MyGroupReducer.class);  
    45.             job.setOutputKeyClass(LongWritable.class);  
    46.             job.setOutputValueClass(LongWritable.class);  
    47.   
    48.             //2.3 指定输出的路径和设置输出的格式化类  
    49.             FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));  
    50.             job.setOutputFormatClass(TextOutputFormat.class);  
    51.   
    52.             // 提交作业 退出  
    53.             System.exit(job.waitForCompletion(true) ? 0 : 1);  
    54.   
    55.         } catch (Exception e) {  
    56.             e.printStackTrace();  
    57.         }  
    58.     }  
    59.   
    60.     public static class MyGroupMapper extends Mapper<LongWritable, Text, CombineKey, LongWritable> {  
    61.         // 创建联合的key  
    62.         private CombineKey combineKey = new CombineKey();  
    63.   
    64.         protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, CombineKey, LongWritable>.Context context) throws IOException,  
    65.                 InterruptedException {  
    66.             // 对输入value进行分割  
    67.             String[] splits = value.toString().split("\t");  
    68.             // 设置联合的Key  
    69.             combineKey.setComKey(Long.parseLong(splits[0]));  
    70.             combineKey.setComVal(Long.parseLong(splits[1]));  
    71.   
    72.             // 写出去  
    73.             context.write(combineKey, new LongWritable(Long.parseLong(splits[1])));  
    74.         }  
    75.     }  
    76.   
    77.     public static class MyGroupReducer extends Reducer<CombineKey, LongWritable, LongWritable, LongWritable> {  
    78.         @Override  
    79.         protected void reduce(CombineKey combineKey, Iterable<LongWritable> values,  
    80.                 Reducer<CombineKey, LongWritable, LongWritable, LongWritable>.Context context) throws IOException, InterruptedException {  
    81.   
    82.             long min = Long.MAX_VALUE;  
    83.             // 遍历比较求出每个组中的最小值  
    84.             for (LongWritable val : values) {  
    85.   
    86.                 if (val.get() < min) {  
    87.                     min = val.get();  
    88.                 }  
    89.             }  
    90.   
    91.             // 把原始数据中的第一列中的元素分组后的组号作为key，所求的最小值为value将结果写出去  
    92.             context.write(new LongWritable(combineKey.getComKey()), new LongWritable(min));  
    93.         }  
    94.     }  
    95. }  
    96.   
    97. /** 
    98.  * 二次排序构造一个新的Key 
    99.  * @author 廖*民 
    100.  * time : 2015年1月18日下午9:15:55 
    101.  * @version 
    102.  */  
    103. class CombineKey implements WritableComparable<CombineKey> {  
    104.   
    105.     private Long comKey;  
    106.     private Long comVal;  
    107.   
    108.     // 无参构造函数必须提供，否则Hadoop的反射机制会报错  
    109.     public CombineKey() {  
    110.     }  
    111.   
    112.     // 有参构造函数  
    113.     public CombineKey(Long comKey, Long comVal) {  
    114.         this.comKey = comKey;  
    115.         this.comVal = comVal;  
    116.     }  
    117.   
    118.     public Long getComKey() {  
    119.         return comKey;  
    120.     }  
    121.   
    122.     public void setComKey(Long comKey) {  
    123.         this.comKey = comKey;  
    124.     }  
    125.   
    126.     public Long getComVal() {  
    127.         return comVal;  
    128.     }  
    129.   
    130.     public void setComVal(Long comVal) {  
    131.         this.comVal = comVal;  
    132.     }  
    133.   
    134.     public void write(DataOutput out) throws IOException {  
    135.         out.writeLong(this.comKey);  
    136.         out.writeLong(this.comVal);  
    137.     }  
    138.   
    139.     public void readFields(DataInput in) throws IOException {  
    140.         this.comKey = in.readLong();  
    141.         this.comVal = in.readLong();  
    142.     }  
    143.   
    144.     /** 
    145.      * 第一列按升序排列,第一列相同时，第二列也按升序排列 
    146.      */  
    147.     public int compareTo(CombineKey o) {  
    148.         long minus = this.comKey - o.comVal;  
    149.         if (minus != 0) {  
    150.             return (int) minus;  
    151.         }  
    152.   
    153.         return (int) (this.comVal - o.comVal);  
    154.     }  
    155.   
    156. }  
    157.   
    158. /** 
    159.  * 自定义分组比较器 
    160.  * @author 廖*民 
    161.  * time : 2015年1月18日下午9:15:26 
    162.  * @version 
    163.  */  
    164. class MyGroupComparator implements RawComparator<CombineKey> {  
    165.   
    166.     // 分组策略中，这个方法不是重点  
    167.     public int compare(CombineKey o1, CombineKey o2) {  
    168.         // TODO Auto-generated method stub  
    169.         return 0;  
    170.     }  
    171.   
    172.     /** 
    173.      * b1 表示第一个参与比较的字节数组 
    174.      * s1 表示第一个字节数组中开始比较的位置  
    175.      * l1 表示第一个字节数组中参与比较的字节长度  
    176.      * b2 表示第二个参与比较的字节数组  
    177.      * s2 表示第二个字节数组中开始比较的位置  
    178.      * l2 表示第二个字节数组参与比较的字节长度 
    179.      */  
    180.     public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {  
    181.   
    182.         // 这里是按第CombineKey中的第一个元素进行分组，因为是long类型，所以是8个字节  
    183.         return WritableComparator.compareBytes(b1, s1, 8, b2, s2, 8);  
    184.     }  
    185.   
    186. }  

MapReduce自定义分组Group

标签：

原文地址：http://www.cnblogs.com/thinkpad/p/5173738.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！