Hadoop MapReduce 学习笔记(十一) MapReduce实现类似SQL的order by/排序3 改进

xiaodouya33

　　本博客属原创文章,转载请注明出处:http://guoyunsky.iyunv.com/blog/1235953
　　请先阅读:           
　　           1.Hadoop MapReduce 学习笔记(一) 序言和准备
　　            2.Hadoop MapReduce 学习笔记(二) 序言和准备 2
　　3.Hadoop MapReduce 学习笔记(八) MapReduce实现类似SQL的order by/排序
　　4.Hadoop MapReduce 学习笔记(九) MapReduce实现类似SQL的order by/排序 正确写法
　　5.Hadoop MapReduce 学习笔记(十) MapReduce实现类似SQL的order by/排序2 对多个字段排序
　　          下一篇:Hadoop MapReduce 学习笔记(十二) MapReduce实现类似SQL的order by/排序3 改进及改正
　　Hadoop MapReduce 学习笔记(十) MapReduce实现类似SQL的order by/排序2 对多个字段排序 可以实现对多个字段的order by/排序.但这里想引入2个参数,因为这2个参数对性能有所提升,所以再补充了这篇博客.具体还是看代码吧.

package com.guoyun.hadoop.mapreduce.study;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.RawComparator;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
* 通过MapReduce实现类似SELECT ＊ FROM TABLE ORDER BY COL1 ASC，COL2 DESC功能
* 也就是对多个字段的排序
* 相比 ＠OrderByMultiMapReduceTest,主要引入了Partitioner和GroupingComparator,提升性能
* 由于生成的数据frameworkName比较固定(具体请查看 @MyMapReduceMultiColumnTest 如何生成的数据)
* 所以这里获取map输出key的frameworkName属性,交给Partitioner和GroupingComparator来确定相同
* frameworkName的数据输出到相同的Reduce上,尽可能减少Reduce之前的清洗和排序工作,提升性能.
* 具体Partitioner和GroupingComparator的用法请查看Hadoop说明
* 这里只是我目前对Partitioner和GroupingComparator的理解,刻意安排的输入数据.一切还需要验证中,待有机会
* 查看map和reduce源码后再来求证.
* 但运行本类后结果和MROutput_Multi_OrderBy并不一致,正确的实现方法请查看类 @OrderByMultiMapReduceImproveFixTest
*/
public class OrderByMultiMapReduceImproveTest extends
OrderByMultiMapReduceTest {
public static final Logger log=LoggerFactory.getLogger(OrderByMultiMapReduceImproveTest.class);
public OrderByMultiMapReduceImproveTest(long dataLength) throws Exception {
super(dataLength);
// TODO Auto-generated constructor stub
}
public OrderByMultiMapReduceImproveTest(String outputPath) throws Exception {
super(outputPath);
// TODO Auto-generated constructor stub
}
public OrderByMultiMapReduceImproveTest(String inputPath, String outputPath) {
super(inputPath, outputPath);
}
public OrderByMultiMapReduceImproveTest(long dataLength, String inputPath,
String outputPath) throws Exception {
super(dataLength, inputPath, outputPath);
// TODO Auto-generated constructor stub
}
/**
* partitioner
*/
public static class MyPartitioner extends Partitioner<OrderMultiColumnWritable,NullWritable>{
@Override
public int getPartition(OrderMultiColumnWritable key, NullWritable value,
int numbers) {
return (int)Math.abs(key.getFrameworkName().hashCode()%numbers);
}
}
public static class MyComparator extends WritableComparator {
public MyComparator() {
super(OrderMultiColumnWritable.class);
}
public int compare(byte[] b1, int s1, int l1,
byte[] b2, int s2, int l2) {
return compareBytes(b1, s1, l1, b2, s2, l2);
}
}
static {                                       
// register this comparator
WritableComparator.define(OrderMultiColumnWritable.class, new MyComparator());
}
/**
* GroupingComparator
*/
public static class MyGroupingComparator implements RawComparator<OrderMultiColumnWritable>{
@Override
public int compare(OrderMultiColumnWritable o1,
OrderMultiColumnWritable o2) {
return o1.getFrameworkName().compareTo(o2.getFrameworkName());
//return o1.compareTo(o2);
}
@Override
public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2,
int l2) {
return WritableComparator.compareBytes(b1,s1,l1,b2,s2,l2);
}
}
public static void main(String[] args){
MyMapReduceTest mapReduceTest=null;
Configuration conf=null;
Job job=null;
FileSystem fs=null;
Path inputPath=null;
Path outputPath=null;
long begin=0;
String input="testDatas/mapreduce/MRInput_Multi_OrderBy";
String output="testDatas/mapreduce/MROutput_Multi_OrderBy_Improve";

try {
// 直接使用MRInput_Single_OrderBy的输入数据,不重新生成数据,以便比对结果是否正确
// 和MROutput_Multi_OrderBy进行比对
mapReduceTest=new OrderByMultiMapReduceImproveTest(input,output);
inputPath=new Path(mapReduceTest.getInputPath());
outputPath=new Path(mapReduceTest.getOutputPath());
conf=new Configuration();
job=new Job(conf,"OrderBy");
fs=FileSystem.getLocal(conf);
if(fs.exists(outputPath)){
if(!fs.delete(outputPath,true)){
System.err.println("Delete output file:"+mapReduceTest.getOutputPath()+" failed!");
return;
}
}

job.setJarByClass(OrderByMultiMapReduceImproveTest.class);
job.setMapOutputKeyClass(OrderMultiColumnWritable.class);
job.setMapOutputValueClass(NullWritable.class);
job.setOutputKeyClass(OrderMultiColumnWritable.class);
job.setOutputValueClass(NullWritable.class);
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class);
job.setPartitionerClass(MyPartitioner.class);
job.setGroupingComparatorClass(MyGroupingComparator.class);
job.setNumReduceTasks(2);
FileInputFormat.addInputPath(job, inputPath);
FileOutputFormat.setOutputPath(job, outputPath);
begin=System.currentTimeMillis();
job.waitForCompletion(true);
System.out.println("===================================================");
if(mapReduceTest.isGenerateDatas()){
System.out.println("The maxValue is:"+mapReduceTest.getMaxValue());
System.out.println("The minValue is:"+mapReduceTest.getMinValue());
}
System.out.println("Spend time:"+(System.currentTimeMillis()-begin));
// Spend time:1280
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

 
更多文章、感悟、分享、勾搭，请用微信扫描: