034 二次排序

一：准备

1.二测排序

　　其中1说明了自定义类型

　　2与3说明了shuffle阶段的分区与分组，同时说明了程序的写法。

　　

2.RawComparator class

　　

3.二次排序的要点

　　组合key，key是一个组合的字段，自定义数据类型

　　　　实现WritableComparable

　　保证原来的分区不变，自定义分区规则

　　　　继承Patitioner

　　保证原来的分组不变，自定义分组规则

　　　　继承RawComparator

4.输入的数据

　　

5.需求

　　平时的只有一次排序，就是第一个会排序，但是输出的结果中第二个没有排序处理。

　　现在希望。在第一个key排序之后，后面的key也可以排序出来。

　　

二：第一次排序

3.输出第一排序的程序

　　MAPPER--------------

　　

　　REDUCER------------

　　

4.结果

　　

三：二次排序

5.map和reduce程序

　　

6.自定义类型的程序

　　需要实现接口WritableComparable

　　输入String，int。

　　

7.自定义分组比较器

　　需要实现RawComparator

　　两个函数都是相同的意思，都是在返回first的比较结果。

　　

8.定义分区规则

　　继承Patitioner

　　

9.运行结果

　　

四：优化点

　　例如分区就属于优化，但是这里说的是正负数的优化。

　　

五：重新整理

1.项目结构

　　

2.程序代码

RealSecondSort.class

package com.senior.sort;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
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.mapred.Partitioner;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.senior.network.WebPvCount;
import com.senior.network.WebPvCount.WebPvCountMapper;
import com.senior.network.WebPvCount.WebPvCountReducer;

public class RealSecondSort extends Configured implements Tool{
	//Mapper
	public static class SortMapper extends Mapper<LongWritable,Text,PariWritable,IntWritable>{
		private PariWritable mapoutkey=new PariWritable();
		private IntWritable mapoutvalue=new IntWritable();
		@Override
		protected void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException {
			String valueStr=value.toString();
			String strs[]=valueStr.split(",");
			mapoutkey.set(strs[0],Integer.valueOf(strs[1]));
			mapoutvalue.set(Integer.valueOf(strs[1]));
			context.write(mapoutkey, mapoutvalue);
		
		}
		
	}
	
	//Reducer
	public static class SortReducer extends Reducer<PariWritable,IntWritable,Text,IntWritable>{
		private Text outkey=new Text();
		@Override
		protected void reduce(PariWritable key, Iterable<IntWritable> value,Context context)throws IOException, InterruptedException {
			for(IntWritable str : value){
				outkey.set(key.getFirst());
				context.write(outkey, str);
			}
		}
		
	}
	
	//Driver
	public int run(String[] args) throws Exception {
		Configuration conf=this.getConf();
		Job job=Job.getInstance(conf,this.getClass().getSimpleName());
		job.setJarByClass(RealSecondSort.class);
		//input
		Path inpath=new Path(args[0]);
		FileInputFormat.addInputPath(job, inpath);
		
		//output
		Path outpath=new Path(args[1]);
		FileOutputFormat.setOutputPath(job, outpath);
		
		//map
		job.setMapperClass(SortMapper.class);
		job.setMapOutputKeyClass(PariWritable.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		//shuffle*************************************
		job.setPartitionerClass(PartitionNum.class);
		job.setGroupingComparatorClass(GroupingComparator.class);	
		
		//shuffle*************************************
		
		//reduce
		job.setReducerClass(SortReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		//submit
		boolean isSucess=job.waitForCompletion(true);
		return isSucess?0:1;

	}
	
	//Main
	public static void main(String[] args) throws Exception{
		Configuration conf=new Configuration();
		args=new String[]{
				"hdfs://linux-hadoop01.ibeifeng.com:8020/user/beifeng/mapreduce/wordcount/inputSortData",
				"hdfs://linux-hadoop01.ibeifeng.com:8020/user/beifeng/mapreduce/wordcount/outputSortData2"
		};
		int status=ToolRunner.run(new RealSecondSort(), args);
		System.exit(status);
	}

}

　

PariWritable.java

这个地方使用的接口可以看看下面的说明。

package com.senior.sort;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.WritableComparable;

public class PariWritable implements WritableComparable<PariWritable>{
	private String first;
	private Integer second;
	public PariWritable(){}
	public PariWritable(String first,Integer second){
		set(first,second);
	}
	//set get
	public String getFirst() {
		return first;
	}
	public void setFirst(String first) {
		this.first = first;
	}
	public Integer getSecond() {
		return second-Integer.MAX_VALUE;
	}
	public void setSecond(Integer second) {
		this.second = second+Integer.MAX_VALUE;
	}
	public void set(String first, Integer second) {
		this.first=first;
		this.second=second;	
	}
	//
	public void readFields(DataInput input) throws IOException {
		this.first=input.readUTF();
		this.second=input.readInt();		
	}
	
	public void write(DataOutput output) throws IOException {
		output.writeUTF(first);
		output.writeInt(second);
		
	}
	public int compareTo(PariWritable o) {
		int comp=this.first.compareTo(o.getFirst());
		if(0!=comp){
			return comp;
		}
		return Integer.valueOf(getSecond()).compareTo(Integer.valueOf(o.getSecond()));
	}
	@Override
	public String toString() {
		return "PariWritable [first=" + first + ", second=" + second + "]";
	}
	
}

　　

PartitionNum.java

package com.senior.sort;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Partitioner;

public class PartitionNum extends Partitioner<PariWritable, IntWritable> {

	@Override
	public int getPartition(PariWritable key, IntWritable value, int num) {
		
		return (key.getFirst().hashCode()&Integer.MAX_VALUE)%num;
	}

}

　　

GroupingComparator.java

关于程序中的一点仔细看下面的一个部分，就可以很好的理解了。

package com.senior.sort;

import org.apache.hadoop.io.RawComparator;
import org.apache.hadoop.io.WritableComparator;

public class GroupingComparator implements RawComparator<PariWritable> {

	public int compare(PariWritable o1, PariWritable o2) {
		return o1.getFirst().compareTo(o2.getFirst());
	}

	public int compare(byte[] b1, int arg1, int l1, byte[] b2, int arg4,int l2) {
		
		return WritableComparator.compareBytes(b1, 0, l1-4, b2, 0, l2-4);
	}

}

　　

3.效果

　　

六：Hadoop的序列化

1.说明

　　在上面的程序中使用到了序列化，在整理的过程中对这一块进行整理一番。

　　

2.序列化的功能　　

　　对于需要保存和处理大规模数据的Hadoop来说，其序列化机制要达到以下目的：

• 排列紧凑：尽量减少带宽，加快数据交换速度
• 处理快速：进程间通信需要大量的数据交互，使用大量的序列化机制，必须减少序列化和反序列的开支
• 跨语言：可以支持不同语言间的数据交互啊，如C++
• 可扩展：当系统协议升级，类定义发生变化，序列化机制需要支持这些升级和变化

3.Writable

　　为了支持以上特性，引用了Writable接口。和说明性Serializable接口不一样，它要求实现两个方法。

　　这个功能已经在前面的章节中说明过。

public interface Writable {  
  void write(DataOutput out) throws IOException;  
  void readFields(DataInput in) throws IOException;  
}  

　　

4.其他接口

　　A。WritableComparable:它不仅提供序列化功能，而且还提供比较的功能。这种比较式基于反序列后的对象成员的值，速度较慢。

　　B。RawComparator:由于MapReduce十分依赖基于键的比较排序（自定义键还需要重写hashCode和equals方法），因此提供了一个优化接口 RawComparator。该接口允许直接比较数据流中的记录，无需把数据流反序列化为对象，这样避免了新建对象的额外开销。RawComparator定义如下，compare方法可以从每个字节数组b1和b2中读取给定起始位置(s1和s2)以及长度l1和l2的一个整数直接进行比较。

　　这个上面已经说明过。

public interface RawComparator<T> extends Comparator<T> {  
  
  public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2);  
  
}  

　　C。WritableComparator: 是 RawComparator 的一个通用实现，提供两个功能：提供了一个 RawComparator的comparea()的默认实现，该默认实现只是反序列化了键然后再比较，没有什么性能优势。其次、充当了 RawComaprator 实例的一个工厂方法。

　　当我们要实现自定key排序时（自定义分组），需要指定自己的排序规则。

　　如需要以StartEndDate为键且以开始时间分组，则需要自定义分组器：

class MyGrouper implements RawComparator<StartEndDate> {  
    @Override  
    public int compare(StartEndDate o1, StartEndDate o2) {  
        return (int)(o1.getStartDate().getTime()- o2.getEndDate().getTime());  
    }  
    @Override  
    public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {  
        int compareBytes = WritableComparator.compareBytes(b1, s1, 8, b2, s2, 8);  
        return compareBytes;  
    }  
       
}