MapReduce 二次排序
发布日期:2021-10-09 07:57:12 浏览次数:1 分类:技术文章

本文共 9106 字,大约阅读时间需要 30 分钟。

默认情况下,Map 输出的结果会对 Key 进行默认的排序,但是有时候需要对 Key 排序的同时再对 Value 进行排序,这时候就要用到二次排序了。下面让我们来介绍一下什么是二次排序。

二次排序原理

        我们把二次排序主要分为以下几个阶段。

        Map 起始阶段

        在Map阶段,使用 job.setInputFormatClass() 定义的 InputFormat ,将输入的数据集分割成小数据块 split,同时 InputFormat 提供一个 RecordReader的实现。本课程中使用的是 TextInputFormat,它提供的 RecordReader 会将文本的行号作为 Key,这一行的文本作为 Value。这就是自定义 Mapper 的输入是 < LongWritable,Text> 的原因。然后调用自定义 Mapper 的map方法,将一个个< LongWritable,Text>键值对输入给 Mapper 的 map方法。

        Map 最后阶段

        在 Map 阶段的最后,会先调用 job.setPartitionerClass() 对这个 Mapper 的输出结果进行分区,每个分区映射到一个Reducer。每个分区内又调用 job.setSortComparatorClass() 设置的 Key 比较函数类排序。可以看到,这本身就是一个二次排序。如果没有通过 job.setSortComparatorClass() 设置 Key 比较函数类,则使用 Key 实现的 compareTo() 方法。我们既可以使用 IntPair 实现的 compareTo() 方法,也可以专门定义 Key 比较函数类。

        Reduce 阶段

        在 Reduce 阶段,reduce() 方法接受所有映射到这个 Reduce 的 map 输出后,也是会调用 job.setSortComparatorClass()方法设置的 Key 比较函数类,对所有数据进行排序。然后开始构造一个 Key 对应的 Value 迭代器。 这时就要用到分组,使用 job.setGroupingComparatorClass()方法设置分组函数类。只要这个比较器比较的两个 Key 相同,它们就属于同一组,它们的 Value 放在一个 Value 迭代器,而这个迭代器的 Key 使用属于同一个组的所有Key的第一个Key。最后就是进入 Reducer 的 reduce() 方法,reduce() 方法的输入是所有的 Key 和它的 Value 迭代器,同样注意输入与输出的类型必须与自定义的 Reducer 中声明的一致。

        接下来我们通过数据示例,可以很直观的了解二次排序的原理。

        输入文件sort.txt()内容为:

40  20 40  1040  3040  530  3030  2030  1030  4050  20 50  5050  1050  60

        输出文件的内容(从小到大排序)如下:

30  1030  2030  3030  40==============================40  540  1040  2040  30==============================  50  1050  2050  5050  60

        从输出的结果可以看出Key实现了从小到大的排序,同时相同Key的Value也实现了从小到大的排序,这就是二次排序的结果。

二次排序的具体流程

        在 MapReduce 中,所有的 Key 是需要被比较和排序的,而且是二次,先根据 Partitioner,再根据大小。而本例中也是要比较两次。先按照第一字段排序,然后再对第一字段相同的按照第二字段排序。根据这一点,我们可以构造一个复合类 IntPair ,它有两个字段,先利用分区对第一字段排序,再利用分区内的比较对第二字段排序。二次排序的流程分为以下几步。

        1、自定义 key

        所有自定义的 key 应该实现接口 WritableComparable,因为它是可序列化的并且可比较的。WritableComparable 的内部方法如下所示。

//反序列化,从流中的二进制转换成IntPairpublic void readFields(DataInput in) throws IOException       //序列化,将IntPair转化成使用流传送的二进制public void write(DataOutput out)//key的比较public int compareTo(IntPair o)       //默认的分区类 HashPartitioner,使用此方法public int hashCode()//默认实现public boolean equals(Object right)

        2、自定义分区

        自定义分区函数类 FirstPartitioner,是 key 的第一次比较,完成对所有 key 的排序。

public static class FirstPartitioner extends Partitioner< IntPair,IntWritable>

        在 job 中使用 setPartitionerClasss()方法设置 Partitioner。

job.setPartitionerClasss(FirstPartitioner.Class);

        3、Key 的比较类

        这是 Key 的第二次比较,对所有的 Key 进行排序,即同时完成IntPair中的first和second排序。该类是一个比较器,可以通过两种方式实现。

        1) 继承 WritableComparator。

public static class KeyComparator extends WritableComparator

        必须有一个构造函数,并且重载 以下方法。

public int compare(WritableComparable w1, WritableComparable w2)

        2) 实现接口 RawComparator。

        上面两种实现方式,在 Job 中,可以通过setSortComparatorClass()方法来设置Key的比较类。

job.setSortComparatorClass(KeyComparator.Class);

        注意:如果没有使用自定义的 SortComparator 类,则默认使用 Key 中compareTo()方法对 Key 排序分组。

        4、定义分组类函数

        在 Reduce 阶段,构造一个与 Key 相对应的 Value 迭代器的时候,只要 first 相同就属于同一个组,放在一个 Value 迭代器。定义这个比较器,可以有两种方式。

        1) 继承 WritableComparator。

public static class GroupingComparator extends WritableComparator

        必须有一个构造函数,并且重载以下方法。

public int compare(WritableComparable w1, WritableComparable w2)

        2) 实现接口 RawComparator。

        上面两种实现方式,在 Job 中,可以通过 setGroupingComparatorClass()方法来设置分组类。

job.setGroupingComparatorClass(GroupingComparator.Class);

        另外注意的是,如果reduce的输入与输出不是同一种类型,则 Combiner和Reducer 不能共用 Reducer 类,因为 Combiner 的输出是 reduce 的输入。除非重新定义一个Combiner。

代码实现

        Hadoop 的 example 包中自带了一个 MapReduce 的二次排序算法,下面这个示例对 example 包中的二次排序源码的改进。我们按照以下几步完成二次排序:

        第一步:自定义IntPair类,将示例数据中的key/value封装成一个整体作为Key,同时实现 WritableComparable 接口并重写其方法。

/*** 自己定义的key类应该实现WritableComparable接口*/public  class IntPair implements WritableComparable
{ int first;//第一个成员变量 int second;//第二个成员变量 public void set(int left, int right){ first = left; second = right; } public int getFirst(){ return first; } public int getSecond(){ return second; } @Override //反序列化,从流中的二进制转换成IntPair public void readFields(DataInput in) throws IOException{ first = in.readInt(); second = in.readInt(); } @Override //序列化,将IntPair转化成使用流传送的二进制 public void write(DataOutput out) throws IOException{ out.writeInt(first); out.writeInt(second); } @Override //key的比较 public int compareTo(IntPair o) { // TODO Auto-generated method stub if (first != o.first){ return first < o.first ? -1 : 1; }else if (second != o.second){ return second < o.second ? -1 : 1; }else{ return 0; } } @Override public int hashCode(){ return first * 157 + second; } @Override public boolean equals(Object right){ if (right == null) return false; if (this == right) return true; if (right instanceof IntPair){ IntPair r = (IntPair) right; return r.first == first && r.second == second; }else{ return false; } }}

        第二步:自定义分区函数类FirstPartitioner,根据 IntPair 中的first实现分区。

/*** 分区函数类。根据first确定Partition。*/public static class FirstPartitioner extends Partitioner
{ @Override public int getPartition(IntPair key, IntWritable value,int numPartitions){ return Math.abs(key.getFirst() * 127) % numPartitions; }}

        第三步:自定义 SortComparator 实现 IntPair 类中的first和second排序。本课程中没有使用这种方法,而是使用 IntPair 中的compareTo()方法实现的。

        第四步:自定义 GroupingComparator 类,实现分区内的数据分组。

/***继承WritableComparator*/public static class GroupingComparator extends WritableComparator{        protected GroupingComparator(){            super(IntPair.class, true);        }        @Override        //Compare two WritableComparables.        public int compare(WritableComparable w1, WritableComparable w2){            IntPair ip1 = (IntPair) w1;            IntPair ip2 = (IntPair) w2;            int l = ip1.getFirst();            int r = ip2.getFirst();            return l == r ? 0 : (l < r ? -1 : 1);        }}

        第五步:编写 MapReduce 主程序实现二次排序。

import java.io.DataInput;import java.io.DataOutput;import java.io.IOException;import java.util.StringTokenizer;import org.apache.hadoop.conf.Configuration;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.io.WritableComparable;import org.apache.hadoop.io.WritableComparator;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.mapreduce.Mapper;import org.apache.hadoop.mapreduce.Partitioner;import org.apache.hadoop.mapreduce.Reducer;import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;public class SecondarySort{    // 自定义map    public static class Map extends Mapper
{ private final IntPair intkey = new IntPair(); private final IntWritable intvalue = new IntWritable(); public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{ String line = value.toString(); StringTokenizer tokenizer = new StringTokenizer(line); int left = 0; int right = 0; if (tokenizer.hasMoreTokens()){ left = Integer.parseInt(tokenizer.nextToken()); if (tokenizer.hasMoreTokens()) right = Integer.parseInt(tokenizer.nextToken()); intkey.set(left, right); intvalue.set(right); context.write(intkey, intvalue); } } } // 自定义reduce public static class Reduce extends Reducer< IntPair, IntWritable, Text, IntWritable>{ private final Text left = new Text(); public void reduce(IntPair key, Iterable< IntWritable> values,Context context) throws IOException, InterruptedException{ left.set(Integer.toString(key.getFirst())); for (IntWritable val : values){ context.write(left, val); } } } /** * @param args */ public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException{ // TODO Auto-generated method stub Configuration conf = new Configuration(); Job job = new Job(conf, "secondarysort"); job.setJarByClass(SecondarySort.class); FileInputFormat.setInputPaths(job, new Path(args[0]));//输入路径 FileOutputFormat.setOutputPath(job, new Path(args[1]));//输出路径 job.setMapperClass(Map.class);// Mapper job.setReducerClass(Reduce.class);// Reducer job.setPartitionerClass(FirstPartitioner.class);// 分区函数 //job.setSortComparatorClass(KeyComparator.Class);//本课程并没有自定义SortComparator,而是使用IntPair自带的排序 job.setGroupingComparatorClass(GroupingComparator.class);// 分组函数 job.setMapOutputKeyClass(IntPair.class); job.setMapOutputValueClass(IntWritable.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); job.setInputFormatClass(TextInputFormat.class); job.setOutputFormatClass(TextOutputFormat.class); System.exit(job.waitForCompletion(true) ? 0 : 1); }}

        至此,MapReduce 的二次排序的原理和实现已经学习完毕,理解起来可能有点难度,希望大家先看代码,把代码跑起来以后再去看前面的原理,这样理解起来就非常容易。

以上就是博主为大家介绍的这一板块的主要内容,这都是博主自己的学习过程,希望能给大家带来一定的指导作用,有用的还望大家点个支持,如果对你没用也望包涵,有错误烦请指出。如有期待可关注博主以第一时间获取更新哦,谢谢! 

 版权声明:本文为博主原创文章,未经博主允许不得转载。

转载地址:https://blog.csdn.net/py_123456/article/details/80270026 如侵犯您的版权,请留言回复原文章的地址,我们会给您删除此文章,给您带来不便请您谅解!

上一篇:Hadoop实战:明星搜索指数统计,找出人气王
下一篇:Hadoop实战:reduce端实现Join

发表评论

最新留言

路过按个爪印,很不错,赞一个!
[***.219.124.196]2024年03月21日 17时28分17秒

关于作者

    喝酒易醉,品茶养心,人生如梦,品茶悟道,何以解忧?唯有杜康!
-- 愿君每日到此一游!

推荐文章

java 减少内存_java中减少内存占用小技巧 2019-04-21
centos 7 mysql图形界面_centos7-vnstat图形界面搭建 2019-04-21
java 防渗透_「java、工程师工作经验怎么写」-看准网 2019-04-21
java中跳出当前循环怎么做_在java中,如何跳出当前的多重循环? 2019-04-21
java程序中执行maven_java – 将一个enviornment变量传递给Maven中的已执行进程 2019-04-21
java16下载_java lombok下载 2019-04-21
python 图像处理与识别书籍_Python图像处理之识别图像中的文字(实例讲解) 2019-04-21
java安全初始化_java安全编码指南之:声明和初始化 2019-04-21
java jstat gc_分析JVM GC及内存情况的方法 2019-04-21
php pclzip.lib.php,php使用pclzip类实现文件压缩的方法(附pclzip类下载地址) 2019-04-21
php dns更新,php_mzdns: 站群,大量域名 通过 dns 服务商 api 批量添加 ip 工具。你懂的~ 基于 mzphp2 框架。... 2019-04-21
jdk 1.8 java.policy,JDK1.8 导致系统报错:java.security.InvalidKeyException:illegal Key Size 2019-04-21
php linux权限,Linux权限详细介绍 2019-04-21
典型环节的matlab仿真分析,典型环节的MATLAB仿真.doc 2019-04-21
Php contenttype类型,各种类型文件的Content Type 2019-04-21
php使用redis持久化,redis如何持久化 2019-04-21
php7.1解压包安装,【Swoole】php7.1安装swoole扩展 2019-04-21
linux centos删除安装的包,CentOS yum认为已删除的软件包仍在安装中 2019-04-21
酒店管理系统c语言带注释,酒店管理系统--C语言版.pdf 2019-04-21
c语言 实现sizeof功能,C语言简单实现sizeof功能代码 2019-04-21