大家好，我是Iggi。
今天我给大家分享的是MapReduce2-3.1.1版本的SecondarySort实验。
关于MapReduce的一段文字简介请自行查阅我的实验示例：MapReduce2-3.1.1 实验示例 单词计数（一）
好，下面进入正题。介绍Java操作MapReduce2组件完成SecondarySort的操作。
首先，使用IDE建立Maven工程，建立工程时没有特殊说明，按照向导提示点击完成即可。重要的是在pom.xml文件中添加依赖包，内容如下图：
image.png
待系统下载好依赖的jar包后便可以编写程序了。
展示实验代码：
package linose.mapreduce.secondarysort;
import java.io.IOException;
import java.io.OutputStreamWriter;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.log4j.BasicConfigurator;
import linose.mapreduce.secondarysort.SecondarySort.FirstPartitioner;
import linose.mapreduce.secondarysort.SecondarySort.KeyComparator;
import linose.mapreduce.secondarysort.SecondarySort.SortMapper;
import linose.mapreduce.secondarysort.SecondarySort.SortReduce;
public class AppSort
{
public static void main( String[] args ) throws IOException, ClassNotFoundException, InterruptedException
{
/**
* 设定MapReduce示例拥有HDFS的操作权限
*/
System.setProperty("HADOOP_USER_NAME", "hdfs");
/**
* 为了清楚的看到输出结果，暂将集群调试信息缺省。
* 如果想查阅集群调试信息，取消注释即可。
*/
BasicConfigurator.configure();
/**
* MapReude实验准备阶段：
* 定义HDFS文件路径
*/
String defaultFS = "hdfs://master2.linose.cloud.beijing.com:8020";
String inputPath = defaultFS + "/index.dirs/inputsort.txt";
String outputPath = defaultFS + "/index.dirs/sort";
/**
* 生产配置，并获取HDFS对象
*/
Configuration conf = new Configuration();
conf.set("fs.defaultFS", defaultFS);
FileSystem system = FileSystem.get(conf);
/**
* 定义输入路径，输出路径
*/
Path inputHdfsPath = new Path(inputPath);
Path outputHdfsPath = new Path(outputPath);
/**
* 如果实验数据文件不存在则创建数据文件
*/
system.delete(inputHdfsPath, false);
if (!system.exists(inputHdfsPath)) {
FSDataOutputStream outputStream = system.create(inputHdfsPath);
OutputStreamWriter file = new OutputStreamWriter(outputStream);
file.write("5\t35\tlee\n");
file.write("11\t21\tAndy\n");
file.write("8\t25\tDa\n");
file.write("4\t23\tCoCo\n");
file.write("9\t21\tAnn\n");
file.write("2\t34\tchap\n");
file.write("10\t45\tYee\n");
file.write("6\t25\tViVi\n");
file.write("1\t33\tIggi\n");
file.write("3\t27\ttony\n");
file.write("7\t29\tsummer\n");
file.close();
outputStream.close();
}
/**
* 如果实验结果目录存在，遍历文件内容全部删除
*/
if (system.exists(outputHdfsPath)) {
RemoteIterator<LocatedFileStatus> fsIterator = system.listFiles(outputHdfsPath, true);
LocatedFileStatus fileStatus;
while (fsIterator.hasNext()) {
fileStatus = fsIterator.next();
system.delete(fileStatus.getPath(), false);
}
system.delete(outputHdfsPath, false);
}
/**
* 创建MapReduce任务并设定Job名称
*/
Job job = Job.getInstance(conf, "Secondary Sort");
job.setJarByClass(SecondarySort.class);
/**
* 设置输入文件、输出文件
*/
FileInputFormat.addInputPath(job, inputHdfsPath);
FileOutputFormat.setOutputPath(job, outputHdfsPath);
/**
* 指定Reduce类输出类型Key类型与Value类型
*/
job.setOutputKeyClass(IntPair.class);
job.setOutputValueClass(NullWritable.class);
/**
* 指定自定义Map类，Reduce类，Partitioner类、SortComparator类。
*/
job.setMapperClass(SortMapper.class);
job.setReducerClass(SortReduce.class);
job.setPartitionerClass(FirstPartitioner.class);
job.setSortComparatorClass(KeyComparator.class);
/**
* 设定Reduce数量并执行
*/
job.setNumReduceTasks(1);
job.waitForCompletion(true);
/**
* 然后轮询进度，直到作业完成。
*/
float progress = 0.0f;
do {
progress = job.setupProgress();
System.out.println("Secondary Sort: 的当前进度：" + progress * 100);
Thread.sleep(1000);
} while (progress != 1.0f && !job.isComplete());
/**
* 如果成功，查看输出文件内容
*/
if (job.isSuccessful()) {
RemoteIterator<LocatedFileStatus> fsIterator = system.listFiles(outputHdfsPath, true);
LocatedFileStatus fileStatus;
while (fsIterator.hasNext()) {
fileStatus = fsIterator.next();
FSDataInputStream outputStream = system.open(fileStatus.getPath());
IOUtils.copyBytes(outputStream, System.out, conf, false);
outputStream.close();
System.out.println("--------------------------------------------");
}
}
}
}
展示MapReduce2-3.1.1组件编写IntPair测试类：
package linose.mapreduce.secondarysort;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.WritableComparable;
public class IntPair implements WritableComparable<IntPair>{
private IntWritable first;
private IntWritable second;
public void set(IntWritable first, IntWritable second) {
this.first = first;
this.second = second;
}
public IntPair() {
set(new IntWritable(), new IntWritable());
}
public IntPair(int first, int second) {
set(new IntWritable(first), new IntWritable(second));
}
public void setFirst(IntWritable first) {
this.first = first;
}
public IntWritable getFirst() {
return first;
}
public void setSecond(IntWritable second) {
this.second = second;
}
public IntWritable getSecond() {
return second;
}
public void readFields(DataInput in) throws IOException {
first.readFields(in);
second.readFields(in);
}
public void write(DataOutput out) throws IOException {
first.write(out);
second.write(out);
}
public int compareTo(IntPair o) {
int compare = first.compareTo(o.first);
if (0 != compare) {
return compare;
}
return second.compareTo(o.second);
}
public int hashCode() {
return first.hashCode()*163+second.hashCode();
}
public boolean equals(Object o) {
if (o instanceof IntPair) {
IntPair pair = (IntPair)o;
return first.equals(pair.first) && second.equals(pair.second);
}
return false;
}
public String toString() {
return first + "\t" + second;
}
}
展示MapReduce2-3.1.1组件编写Secondary Sort测试类：
package linose.mapreduce.secondarysort;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
public class SecondarySort {
public static class SortMapper extends Mapper<LongWritable, Text, IntPair, NullWritable> {
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String[] fields = value.toString().split("\t");
int field1 = Integer.parseInt(fields[0]);
int field2 = Integer.parseInt(fields[1]);
context.write(new IntPair(field1, field2), NullWritable.get());
}
}
public static class SortReduce extends Reducer<IntPair, NullWritable, IntPair, NullWritable> {
protected void reduce(IntPair key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
context.write(key, NullWritable.get());
}
}
public static class FirstPartitioner extends Partitioner<IntPair, NullWritable> {
public int getPartition(IntPair key, NullWritable value, int partitions) {
return Math.abs(key.getFirst().get()) % partitions;
}
}
public static class KeyComparator extends WritableComparator {
protected KeyComparator() {
super(IntPair.class, true);
}
public int compare(@SuppressWarnings("rawtypes") WritableComparable value1, @SuppressWarnings("rawtypes") WritableComparable value2) {
IntPair pair1 = (IntPair)value1;
IntPair pair2 = (IntPair)value2;
int compare = pair1.getFirst().compareTo(pair2.getFirst());
if (0 != compare) {
return compare;
}
return -pair1.getSecond().compareTo(pair2.getSecond());
}
}
}
下图为测试结果：
image.png
至此，MapReduce2-3.1.1 Secondary Sort 实验示例演示完毕。
