在直接学习hadoop的排序之前还要了解一些基本知识。
Hadoop的序列化和比较接口
Hadoop的序列化格式：Writable
Writable是Hadoop自己的序列化格式，还要一个子接口是WritableComparable<T>，
public interface WritableComparable<T> extends Writable, Comparable<T>
这样一来WritableComparable接口不仅有序列化的功能，还可以进行比较。
排序在MapReduce是很重要的一个方面，因为MapReduce有一个基于键的排序过程，所以可以作为键的类型必须具有Comparable<T>的特性。
除了WritableComparable接口之外，还有一个叫做RawComparator的接口。
WritableComparable和RawComparator的区别是：
WritableComparable需要把数据反序列化为对象，然后做对象之间的比较；RawComparator是直接比较数据流的数据，不需要数据反序列化成对象，从而省去了新建对象的开销。
Hadoop中key的排序逻辑
Hadoop中key的数据类型的排序逻辑其实就是按照WritableComparable<T>的基本数据类型和其他类型的compareTo方法的定义。
Key的排序规则：
- 如果调用了jobconf的setOutputKeyComparatorClass()，那么就使用设置的；
- 否则，采用实现接口WritableComparable的compareTo()来比较大小并排序。
例如IntWritable的比较算法
public int compareTo(Object o) {
int thisValue = this.value;
int thatValue = ((IntWritable)o).value;
return (thisValue<thatValue ? -1 : (thisValue==thatValue ? 0 : 1));
}
我们可以根据自己的需要来修改compareTo()这个方法来实现自己的比价算法。
在Hadoop内部是自动调用这个方法来进行比较的，不需要我们手动干预，而且排序只是局限于map或者reduce内部。针对于map和map之间，reduce与reduce之间的排序compareTo就管不着了。虽然这种情况不常出现，但是还是有，比如全排序。
全排序
如果遇到了全排序，就要关注Partition这个阶段，Partition阶段是针对每个Reduce创建一个分区，然后把Map的输出结果映射到特定的分区中，这个分区中可能会有N个Key对应的数据，但是一个Key只能出现在一个分区中。在实现全排序的过程中，如果只有一个Reduce，也就是只有一个Partition，那么所有Map的输出都会经过一个Partition到一个reduce里，一个reduce里就可以根据compareTo来排序，这样就实现了全排序。但是这样不好的地方就是你全排序都是在一个Reduce中做的，这还叫毛的分布式计算啊。
因此，为了take advantage of distributed computing，我们需要使用多个Reduce，也就是有多个分区，而且这些分区的特点是两个已经排序完毕的partition1和partition2，partition1中的根据排序规则要么全部大于partition2，要么全部小于partition2。
这样做可以利用到分布式计算的特性，但是仍然要考虑到的是分区不均的问题，也就是一个partition中的数要远比其他的partition中的多，这样的话对分布式计算的效果也会大打折扣。解决这个问题的方法就是抽样，目的就是使partition更加均匀。
二次排序
代码实例
需求：现在有这样类型的文件若干
file1.txt
2
32
654
32
15
756
65223
file2.txt
5956
22
650
92
file3.txt
26
54
6
现在希望你用Hadoop进行处理并得到如下结果：
1 2
2 6
3 15
4 22
5 26
6 32
7 32
8 54
9 92
10 650
11 654
12 756
13 5956
14 65223
注：本代码是在Hadoop 2.7.4真实分布式环境下执行通过的。
当然最简单的就是多个Map，一个Reduce；因为Reduce在处理的时候是默认的会排序的。所以可以得到一个最简单的版本。
Version1：最简单版本，多个Map，单个Reduce
package com.tuhooo; import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.lib.partition.InputSampler;
import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser; public class Sort { // map将输入中的value化成IntWritable类型, 作为输出的key
public static class Map extends Mapper<Object, Text, IntWritable, IntWritable> { private static IntWritable data = new IntWritable();
// 实现map函数
@Override
public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
String line = value.toString();
if(line != null && line.length() != 0 && !"".equals(line)) {
data.set(Integer.parseInt(line));
context.write(data, new IntWritable(1));
}
}
} // reduce会对map传进来的值进行排序
public static class Reduce extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> { private static IntWritable linenum = new IntWritable(1); @Override
public void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
for(IntWritable val : values) {
context.write(linenum, key);
linenum = new IntWritable(linenum.get() + 1);
}
}
} public static void main(String[] args) throws Exception { // 校验参数是否正确
if (args.length != 2) {
System.err.println("Usage: MaxTemperature <input path> <output path>");
System.exit(-1);
} Job job = new Job();
job.setJarByClass(Sort.class); // main Class
job.setJobName("Sort"); FileInputFormat.addInputPath(job, new Path(args[0])); // 输入数据的目录
FileOutputFormat.setOutputPath(job, new Path(args[1])); // 输出数据的目录 job.setMapperClass(Map.class);
job.setReducerClass(Reduce.class);
job.setOutputKeyClass(IntWritable.class);
job.setOutputValueClass(IntWritable.class);
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
通过以上的讲解我们知道这得分区啊，于是有了下面的版本
Version2：M个Map，N个Reduce，N个Partition，均值分区不采样
package com.tuhooo; import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.lib.partition.InputSampler;
import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser; public class Sort { // map将输入中的value化成IntWritable类型, 作为输出的key
public static class Map extends Mapper<Object, Text, IntWritable, IntWritable> { private static IntWritable data = new IntWritable();
// 实现map函数
@Override
public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
String line = value.toString();
if(line != null && line.length() != 0 && !"".equals(line)) {
data.set(Integer.parseInt(line));
context.write(data, new IntWritable(1));
}
}
} // reduce会对map传进来的值进行排序
public static class Reduce extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> { private static IntWritable linenum = new IntWritable(1); @Override
public void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
for(IntWritable val : values) {
context.write(linenum, key);
linenum = new IntWritable(linenum.get() + 1);
}
}
} public static class MyPartition extends Partitioner<IntWritable, IntWritable> { @Override
public int getPartition(IntWritable key, IntWritable value, int numTaskReduce) { int maxNumber = 65223; // 样本数据中的最大值
int part = maxNumber/numTaskReduce + 1;
int keyNum = key.get();
for(int i=0; i<numTaskReduce; i++) { if(keyNum >= part*i && keyNum <= part*(i+1)) {
return i;
}
}
return -1; // 如果没有出现在分区里面就会返回-1, 如果返回-1肯定是要报错的
}
}
public static void main(String[] args) throws Exception { // 校验参数是否正确
if (args.length != 2) {
System.err.println("Usage: MaxTemperature <input path> <output path>");
System.exit(-1);
}
Job job = new Job();
job.setJarByClass(Sort.class); // main Class
job.setJobName("Sort"); FileInputFormat.addInputPath(job, new Path(args[0])); // 输入数据的目录
FileOutputFormat.setOutputPath(job, new Path(args[1])); // 输出数据的目录 job.setNumReduceTasks(3); // Reduce任务数为3
job.setPartitionerClass(MyPartition.class); // 设置分区的类 job.setMapperClass(Map.class);
job.setReducerClass(Reduce.class);
job.setOutputKeyClass(IntWritable.class);
job.setOutputValueClass(IntWritable.class);
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
Version2的代码基本上可以实现分区排序的功能，但是并没有采样啊，就可能会导致分区不均匀导致有些task任务很重，所以我们要加入采样。因此有了Version3。
Version3：M个Map，N个Reduce，N个Partition（此时采用的是TotalSortPartitioner，并且随机采样）
package com.tuhooo; import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.lib.partition.InputSampler;
import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser; public class Sort { // map将输入中的value化成LongWritable类型, 作为输出的key
public static class Map extends Mapper<Object, Text, LongWritable, LongWritable> { private static LongWritable data = new LongWritable();
// 实现map函数
@Override
public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
String line = value.toString();
if(line != null && line.length() != 0 && !"".equals(line)) {
data.set(Integer.parseInt(line));
context.write(data, new LongWritable(1));
}
}
} public static class Reduce extends Reducer<LongWritable, LongWritable, LongWritable, LongWritable> { private static LongWritable linenum = new LongWritable(1); @Override
public void reduce(LongWritable key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
for(LongWritable val : values) {
context.write(linenum, key);
linenum = new LongWritable(linenum.get() + 1);
}
}
} public static void main(String[] args) throws Exception { if (args.length != 3) {
System.err.println("Usage: MaxTemperature <input path> <output path> <partition path>");
System.exit(-1);
} Job job = new Job();
job.setJarByClass(Sort.class); // main Class
job.setJobName("Sort"); job.setNumReduceTasks(3); FileInputFormat.addInputPath(job, new Path(args[0])); // 输入数据的目录
FileOutputFormat.setOutputPath(job, new Path(args[1])); // 输出数据的目录 TotalOrderPartitioner.setPartitionFile(job.getConfiguration(), new Path(args[2]));
InputSampler.Sampler<LongWritable, LongWritable> sampler = new InputSampler.RandomSampler<LongWritable, LongWritable>(0.5, 5, 2);
InputSampler.writePartitionFile(job, sampler); job.setMapperClass(Map.class);
job.setReducerClass(Reduce.class); // 这里换成了TotalOrderPartitioner
job.setPartitionerClass(TotalOrderPartitioner.class); // job.setPartitionerClass(MyPartition.class);
job.setOutputKeyClass(LongWritable.class);
job.setOutputValueClass(LongWritable.class);
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
在编写并运行Version3的时候踩了很多坑：
1. 随机采样的参数设置问题，由于我的样本文件中的数据很少，因此我猜测采样出来的数也很少，导致报异常：数组越界，
此时我采用的参数为：
InputSampler.Sampler<LongWritable, LongWritable> sampler = new InputSampler.RandomSampler<LongWritable, LongWritable>(0.01, 500, 10);
我感觉是没有采到样导致的。
17/11/16 17:31:51 INFO input.FileInputFormat: Total input paths to process : 3
17/11/16 17:31:51 INFO partition.InputSampler: Using 0 samples
17/11/16 17:31:51 INFO zlib.ZlibFactory: Successfully loaded & initialized native-zlib library
17/11/16 17:31:51 INFO compress.CodecPool: Got brand-new compressor [.deflate]
Exception in thread "main" java.lang.ArrayIndexOutOfBoundsException: 0
at org.apache.hadoop.mapreduce.lib.partition.InputSampler.writePartitionFile(InputSampler.java:340)
at com.tuhooo.Sort.main(Sort.java:67)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.hadoop.util.RunJar.run(RunJar.java:221)
at org.apache.hadoop.util.RunJar.main(RunJar.java:136)
2. 如果你有发现的话，最开始在代码Version1和代码Version2中，我用的都是IntWritable，但是在代码Version3中我用的是LongWritable，这是为啥捏。如果我继续在Version3中使用IntWritable的时候，采样文件的结果是这样的：
[root@hadoop Sort]# hdfs dfs -cat /partitionFile_heheda
SEQ!org.apache.hadoop.io.LongWritable!org.apache.hadoop.io.NullWritable*org.apache.hadoop.io.compress.DefaultCodec▒]▒S▒3n▒l▒▒▒▒Axx▒[root@hadoop Sort]#
后来一直报错就是IntWritable不是LongWritable，就是下面这样的：
Error: java.lang.IllegalArgumentException: Can't read partitions file
at org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner.setConf(TotalOrderPartitioner.java:116)
at org.apache.hadoop.util.ReflectionUtils.setConf(ReflectionUtils.java:76)
at org.apache.hadoop.util.ReflectionUtils.newInstance(ReflectionUtils.java:136)
at org.apache.hadoop.mapred.MapTask$NewOutputCollector.<init>(MapTask.java:702)
at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:770)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:341)
at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:164)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1746)
at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:158)
Caused by: java.io.IOException: wrong key class: org.apache.hadoop.io.IntWritable is not class org.apache.hadoop.io.LongWritable
at org.apache.hadoop.io.SequenceFile$Reader.next(SequenceFile.java:2329)
at org.apache.hadoop.io.SequenceFile$Reader.next(SequenceFile.java:2381)
at org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner.readPartitions(TotalOrderPartitioner.java:306)
at org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner.setConf(TotalOrderPartitioner.java:88)
... 10 more
大数据路长漫漫~~~