本文主要介绍下二次排序的实现方式
我们知道MapReduce是按照key来进行排序的,那么如果有个需求就是先按照第一个字段排序,在第一个字段相等的情况下,按照第二个字段排序,这就是传说中的二次排序。
下面就具体说一下二次排序的实现方式
主要就是4点
1.自定义一个Key
为什么要自定义一个Key,我们知道MapReduce中排序就是按照Key来排序的,我们既然想要实现按照两个字段进行排序,默认的方式肯定是不行的,所以自定义一个新的Key,Key里面有两个属性,也就是我们要排序的两个字段。
首先,实现WritableComparable接口,因为Key是可序列化并且可以比较的。其次,重载相关的方法,例如序列化、反序列化相关的方法write()、readFields()。重载在分区的时候要用到的hashCode()方法,注意后面会说到一个Partitioner类,也是用来分区的,用hashCode()方法和Partitioner类进行分区都是可以的,使用其中的一个即可。重载排序用的compareTo()方法。
2.分区函数类
上面定义了一个新的Key,那么我现在做分发,到底按照什么样的规则进行分发是在分区函数中定义的,这个类要继承Partitioner类,重载其中的分区方法getPartition(),在main()函数里面给job添加上即可,例如:job.setPartitionerClass(XXX.class);
注:这个类的作用和新Key中的hashCode()方法作用一样,所以如果在新key的hashCode()方法中写了分区的实现,这个分区类是可以省略的。
3.比较函数类
这个类决定着Key的排序规则,是一个比较器,需要继承WritableComparator类,并且重载其中的compare()方法。在main()函数里给job添加上即可,例如:job.setSortComparatorClass(XXX.class);
注:这个类的作用跟自定义Key的compareTo()方法一样,如果在自定义的Key中重载了compareTo()方法,这个类是可以省略的。
4.分组函数类
通过分区类,我们重新定义了key的分区规则,但是多个key不同的也可以进入一个reduce中,这不是我们想要的,我们需要分区函数来定义什么样的key可以进入相应的reduce来执行,因为也涉及到比较,所以这个类也需要继承WritableComparator,也可以实现RawComparator,并且重载其中的compare()方法,在main()函数中给job加上即可,如:job.setGroupingComparatorClass(XXX.class)。
下面我们来重新简化一下上一篇文章中提到了例子:
#需求:第一列升序,第一列相同时,第二列升序,其第一列相同的放在一个分区中输出
sort1 1 sort2 3 sort2 88 sort2 54 sort1 2 sort6 22 sort6 888 sort6 58
#预期输出结果
#part-r-00000文件
sort1 1 sort1 2
#part-r-00001文件
sort2 3 sort2 54 sort2 88
#part-r-00002
sort6 22 sort6 58 sort6 888
1.自定义组合键
public class CombinationKey implements WritableComparable<CombinationKey>{ private Text firstKey; private IntWritable secondKey; //无参构造函数 public CombinationKey() { this.firstKey = new Text(); this.secondKey = new IntWritable(); } //有参构造函数 public CombinationKey(Text firstKey, IntWritable secondKey) { this.firstKey = firstKey; this.secondKey = secondKey; } public Text getFirstKey() { return firstKey; } public void setFirstKey(Text firstKey) { this.firstKey = firstKey; } public IntWritable getSecondKey() { return secondKey; } public void setSecondKey(IntWritable secondKey) { this.secondKey = secondKey; } public void write(DataOutput out) throws IOException { this.firstKey.write(out); this.secondKey.write(out); } public void readFields(DataInput in) throws IOException { this.firstKey.readFields(in); this.secondKey.readFields(in); } /** * 自定义比较策略 * 注意:该比较策略用于MapReduce的第一次默认排序 * 也就是发生在Map端的sort阶段 * 发生地点为环形缓冲区(可以通过io.sort.mb进行大小调整) */ public int compareTo(CombinationKey combinationKey) { int minus = this.getFirstKey().compareTo(combinationKey.getFirstKey()); if (minus != 0){ return minus; } return this.getSecondKey().get() - combinationKey.getSecondKey().get(); } /* public int compareTo(CombinationKey combinationKey) { System.out.println("------------------------CombineKey flag-------------------"); return this.firstKey.compareTo(combinationKey.getFirstKey()); }*/ @Override public int hashCode() { final int prime = 31; int result = 1; result = prime * result + ((firstKey == null) ? 0 : firstKey.hashCode()); return result; } @Override public boolean equals(Object obj) { if (this == obj) return true; if (obj == null) return false; if (getClass() != obj.getClass()) return false; CombinationKey other = (CombinationKey) obj; if (firstKey == null) { if (other.firstKey != null) return false; } else if (!firstKey.equals(other.firstKey)) return false; return true; } }
2.自定义分组
/** * 自定义分组有中方式,一种是继承WritableComparator * 另外一种是实现RawComparator接口 * @author 廖钟民 * time : 2015年1月19日下午3:30:11 * @version */ public class DefinedGroupSort extends WritableComparator{ protected DefinedGroupSort() { super(CombinationKey.class,true); } @Override public int compare(WritableComparable a, WritableComparable b) { System.out.println("---------------------进入自定义分组---------------------"); CombinationKey combinationKey1 = (CombinationKey) a; CombinationKey combinationKey2 = (CombinationKey) b; System.out.println("---------------------分组结果:" + combinationKey1.getFirstKey().compareTo(combinationKey2.getFirstKey())); System.out.println("---------------------结束自定义分组---------------------"); //自定义按原始数据中第一个key分组 return combinationKey1.getFirstKey().compareTo(combinationKey2.getFirstKey()); } }
3.自定义分区
/** * 自定义分区 * * @author 廖钟*民 time : 2015年1月19日下午12:13:54 * @version */ public class DefinedPartition extends Partitioner<CombinationKey, IntWritable> { /** * 数据输入来源:map输出 我们这里根据组合键的第一个值作为分区 如果不自定义分区的话,MapReduce会根据默认的Hash分区方法 将整个组合键相等的分到一个分区中,这样的话显然不是我们要的效果 * * @param key * map输出键值 * @param value * map输出value值 * @param numPartitions * 分区总数,即reduce task个数 */ public int getPartition(CombinationKey key, IntWritable value, int numPartitions) { if (key.getFirstKey().toString().endsWith("1")){ return 0; } else if (key.getFirstKey().toString().endsWith("2")){ return 1; } else { return 2; } } }
4.主类
public class SecondSortMapReduce extends Configured implements Tool{ // 定义输入路径 private String INPUT_PATH = ""; // 定义输出路径 private String OUT_PATH = ""; public static void main(String[] args) { try { ToolRunner.run(new SecondSortMapReduce(), args); } catch (Exception e) { e.printStackTrace(); } } public static class SecondSortMapper extends Mapper<Text, Text, CombinationKey, IntWritable>{ /** * 这里要特殊说明一下,为什么要将这些变量写在map函数外边 * 对于分布式的程序,我们一定要注意到内存的使用情况,对于MapReduce框架 * 每一行的原始记录的处理都要调用一次map()函数,假设,这个map()函数要处理1一亿 * 条输入记录,如果将这些变量都定义在map函数里面则会导致这4个变量的对象句柄 * 非常的多(极端情况下将产生4*1亿个句柄,当然java也是有自动的GC机制的,一定不会达到这么多) * 导致栈内存被浪费掉,我们将其写在map函数外面,顶多就只有4个对象句柄 */ private CombinationKey combinationKey = new CombinationKey(); Text sortName = new Text(); IntWritable score = new IntWritable(); String[] splits = null; protected void map(Text key, Text value, Mapper<Text, Text, CombinationKey, IntWritable>.Context context) throws IOException, InterruptedException { System.out.println("---------------------进入map()函数---------------------"); //过滤非法记录(这里用计数器比较好) if (key == null || value == null || key.toString().equals("")){ return; } //构造相关属性 sortName.set(key.toString()); score.set(Integer.parseInt(value.toString())); //设置联合key combinationKey.setFirstKey(sortName); combinationKey.setSecondKey(score); //通过context把map处理后的结果输出 context.write(combinationKey, score); System.out.println("---------------------结束map()函数---------------------"); } } public static class SecondSortReducer extends Reducer<CombinationKey, IntWritable, Text, Text>{ StringBuffer sb = new StringBuffer(); Text score = new Text(); /** * 这里要注意一下reduce的调用时机和次数: * reduce每次处理一个分组的时候会调用一次reduce函数。 * 所谓的分组就是将相同的key对应的value放在一个集合中 * 例如:<sort1,1> <sort1,2> * 分组后的结果就是 * <sort1,{1,2}>这个分组会调用一次reduce函数 */ protected void reduce(CombinationKey key, Iterable<IntWritable> values, Reducer<CombinationKey, IntWritable, Text, Text>.Context context) throws IOException, InterruptedException { //将联合Key的第一个元素作为新的key,遍历values将结果写出去 for (IntWritable val : values){ context.write(key.getFirstKey(), new Text(String.valueOf(val.get()))); } } } public int run(String[] args) throws Exception { // 给路径赋值 INPUT_PATH = args[0]; OUT_PATH = args[1]; try { // 创建配置信息 Configuration conf = new Configuration(); conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, "\t"); // 创建文件系统 FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), conf); // 如果输出目录存在,我们就删除 if (fileSystem.exists(new Path(OUT_PATH))) { fileSystem.delete(new Path(OUT_PATH), true); } // 创建任务 Job job = new Job(conf, SecondSortMapReduce.class.getName()); //设置成jar运行型 job.setJarByClass(SecondSortMapReduce.class); // 1.1 设置输入目录和设置输入数据格式化的类 FileInputFormat.setInputPaths(job, INPUT_PATH); job.setInputFormatClass(KeyValueTextInputFormat.class); // 部1.2 设置自定义Mapper类和设置map函数输出数据的key和value的类型 job.setMapperClass(SecondSortMapper.class); job.setMapOutputKeyClass(CombinationKey.class); job.setMapOutputValueClass(IntWritable.class); // 1.3 设置分区和reduce数量(reduce的数量,和分区的数量对应) job.setPartitionerClass(DefinedPartition.class); job.setNumReduceTasks(3); //设置自定义分组策略 job.setGroupingComparatorClass(DefinedGroupSort.class); //设置自定义比较策略(因为我的CombineKey重写了compareTo方法,所以这个可以省略) //job.setSortComparatorClass(DefinedComparator.class); // 1.4 排序 //1.5 归约 // 2.1 Shuffle把数据从Map端拷贝到Reduce端。 // 2.2 指定Reducer类和输出key和value的类型 job.setReducerClass(SecondSortReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); //2.3 指定输出的路径和设置输出的格式化类 FileOutputFormat.setOutputPath(job, new Path(OUT_PATH)); job.setOutputFormatClass(TextOutputFormat.class); // 提交作业 退出 System.exit(job.waitForCompletion(true) ? 0 : 1); } catch (Exception e) { e.printStackTrace(); } return 0; } }
注:使用jar包的方式运行程序一定不要忘记很重要的一句:job.setJarByClass(XXX);
程序运行的结果: