大家好，我是Iggi。
今天我給大家分享的是MapReduce2-3.1.1版本的SecondarySort實(shí)驗(yàn)。
關(guān)于MapReduce的一段文字簡(jiǎn)介請(qǐng)自行查閱我的實(shí)驗(yàn)示例：MapReduce2-3.1.1 實(shí)驗(yàn)示例 單詞計(jì)數(shù)（一）
好，下面進(jìn)入正題。介紹Java操作MapReduce2組件完成SecondarySort的操作。
首先，使用IDE建立Maven工程，建立工程時(shí)沒(méi)有特殊說(shuō)明，按照向?qū)崾军c(diǎn)擊完成即可。重要的是在pom.xml文件中添加依賴(lài)包，內(nèi)容如下圖：
image.png
待系統(tǒng)下載好依賴(lài)的jar包后便可以編寫(xiě)程序了。
展示實(shí)驗(yàn)代碼：
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
{
/**
* 設(shè)定MapReduce示例擁有HDFS的操作權(quán)限
*/
System.setProperty("HADOOP_USER_NAME", "hdfs");
/**
* 為了清楚的看到輸出結(jié)果，暫將集群調(diào)試信息缺省。
* 如果想查閱集群調(diào)試信息，取消注釋即可。
*/
BasicConfigurator.configure();
/**
* MapReude實(shí)驗(yàn)準(zhǔn)備階段：
* 定義HDFS文件路徑
*/
String defaultFS = "hdfs://master2.linose.cloud.beijing.com:8020";
String inputPath = defaultFS + "/index.dirs/inputsort.txt";
String outputPath = defaultFS + "/index.dirs/sort";
/**
* 生產(chǎn)配置，并獲取HDFS對(duì)象
*/
Configuration conf = new Configuration();
conf.set("fs.defaultFS", defaultFS);
FileSystem system = FileSystem.get(conf);
/**
* 定義輸入路徑，輸出路徑
*/
Path inputHdfsPath = new Path(inputPath);
Path outputHdfsPath = new Path(outputPath);
/**
* 如果實(shí)驗(yàn)數(shù)據(jù)文件不存在則創(chuàng)建數(shù)據(jù)文件
*/
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();
}
/**
* 如果實(shí)驗(yàn)結(jié)果目錄存在，遍歷文件內(nèi)容全部刪除
*/
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);
}
/**
* 創(chuàng)建MapReduce任務(wù)并設(shè)定Job名稱(chēng)
*/
Job job = Job.getInstance(conf, "Secondary Sort");
job.setJarByClass(SecondarySort.class);
/**
* 設(shè)置輸入文件、輸出文件
*/
FileInputFormat.addInputPath(job, inputHdfsPath);
FileOutputFormat.setOutputPath(job, outputHdfsPath);
/**
* 指定Reduce類(lèi)輸出類(lèi)型Key類(lèi)型與Value類(lèi)型
*/
job.setOutputKeyClass(IntPair.class);
job.setOutputValueClass(NullWritable.class);
/**
* 指定自定義Map類(lèi)，Reduce類(lèi)，Partitioner類(lèi)、SortComparator類(lèi)。
*/
job.setMapperClass(SortMapper.class);
job.setReducerClass(SortReduce.class);
job.setPartitionerClass(FirstPartitioner.class);
job.setSortComparatorClass(KeyComparator.class);
/**
* 設(shè)定Reduce數(shù)量并執(zhí)行
*/
job.setNumReduceTasks(1);
job.waitForCompletion(true);
/**
* 然后輪詢(xún)進(jìn)度，直到作業(yè)完成。
*/
float progress = 0.0f;
do {
progress = job.setupProgress();
System.out.println("Secondary Sort: 的當(dāng)前進(jìn)度：" + progress * 100);
Thread.sleep(1000);
} while (progress != 1.0f && !job.isComplete());
/**
* 如果成功，查看輸出文件內(nèi)容
*/
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組件編寫(xiě)IntPair測(cè)試類(lèi)：
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組件編寫(xiě)Secondary Sort測(cè)試類(lèi)：
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());
}
}
}
下圖為測(cè)試結(jié)果：
image.png
至此，MapReduce2-3.1.1 Secondary Sort 實(shí)驗(yàn)示例演示完畢。
