2018-07-30期 MapReduce對象排序（單列排序）
package cn.sjq.mr.sort;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.WritableComparable;
/**
* 定義一個員工薪水的JavaBean，並實現MapReduce的WritableComparable<EmpSalaryBean>序列化及比較器接口
* @author songjq
*
*/
public class EmpSalaryBean1 implements WritableComparable<EmpSalaryBean1> {
/*
定義成員屬性
c_oid
c_employee_name
c_second_company_name
c_third_company_name
c_fourth_company_name
c_company_name
c_dept_name
c_sub_total
c_com_fund_payamt
*/
private int seq;
private String c_oid;
private String c_employee_name;
private String c_second_company_name;
private String c_third_company_name;
private String c_fourth_company_name;
private String c_company_name;
private String c_dept_name;
private float c_sub_total;
private float c_com_fund_payamt;
public int getSeq() {
return seq;
}
public void setSeq(int seq) {
this.seq = seq;
}
public String getC_oid() {
return c_oid;
}
public void setC_oid(String c_oid) {
this.c_oid = c_oid;
}
public String getC_employee_name() {
return c_employee_name;
}
public void setC_employee_name(String c_employee_name) {
this.c_employee_name = c_employee_name;
}
public String getC_second_company_name() {
return c_second_company_name;
}
public void setC_second_company_name(String c_second_company_name) {
this.c_second_company_name = c_second_company_name;
}
public String getC_third_company_name() {
return c_third_company_name;
}
public void setC_third_company_name(String c_third_company_name) {
this.c_third_company_name = c_third_company_name;
}
public String getC_fourth_company_name() {
return c_fourth_company_name;
}
public void setC_fourth_company_name(String c_fourth_company_name) {
this.c_fourth_company_name = c_fourth_company_name;
}
public String getC_company_name() {
return c_company_name;
}
public void setC_company_name(String c_company_name) {
this.c_company_name = c_company_name;
}
public String getC_dept_name() {
return c_dept_name;
}
public void setC_dept_name(String c_dept_name) {
this.c_dept_name = c_dept_name;
}
public float getC_sub_total() {
return c_sub_total;
}
public void setC_sub_total(float c_sub_total) {
this.c_sub_total = c_sub_total;
}
public float getC_com_fund_payamt() {
return c_com_fund_payamt;
}
public void setC_com_fund_payamt(float c_com_fund_payamt) {
this.c_com_fund_payamt = c_com_fund_payamt;
}
//方序列化方法
@Override
public void readFields(DataInput in) throws IOException {
this.seq = in.readInt();
this.c_oid = in.readUTF();
this.c_employee_name = in.readUTF();
this.c_second_company_name = in.readUTF();
this.c_third_company_name = in.readUTF();
this.c_fourth_company_name = in.readUTF();
this.c_company_name = in.readUTF();
this.c_dept_name = in.readUTF();
this.c_sub_total = in.readFloat();
this.c_com_fund_payamt = in.readFloat();
}
//序列化方法
@Override
public void write(DataOutput out) throws IOException {
out.writeInt(this.seq);
out.writeUTF(this.c_oid);
out.writeUTF(this.c_employee_name);
out.writeUTF(this.c_second_company_name);
out.writeUTF(this.c_third_company_name);
out.writeUTF(this.c_fourth_company_name);
out.writeUTF(this.c_company_name);
out.writeUTF(this.c_dept_name);
out.writeFloat(this.c_sub_total);
out.writeFloat(this.c_com_fund_payamt);
}
@Override
public String toString() {
return this.seq+"\t"+this.c_oid+"\t"+
this.c_employee_name+"\t"+this.c_second_company_name+"\t"+
this.c_third_company_name+"\t"+this.c_fourth_company_name+"\t"+
this.c_company_name+"\t"+this.c_dept_name+"\t"+
this.c_sub_total+"\t"+this.c_com_fund_payamt+"\t";
}
/*
* 實現compareTo方法
* (non-Javadoc)
* @see java.lang.Comparable#compareTo(java.lang.Object)
*/
@Override
public int compareTo(EmpSalaryBean1 o) {
// 單列排序，這裏根據部門名稱c_dept_name進行升序排序，部門號為字符串
if (this.c_dept_name.compareTo(o.getC_dept_name()) >= 0) {
// 當前c_dept_name大於等於傳入的c_dept_name，則返回1
return 1;
} else {
// 當前c_dept_name小於傳入的c_dept_name，則返回-1
return -1;
}
}
}
2、ObjectSortSingleClumn實現類
package cn.sjq.mr.sort;
import java.io.IOException;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
/**
* 利用MapReduce對EmpSalaryBean排序
* @author songjq
*
*/
public class ObjectSortSingleClumn {
/**
* 排序Mapper類
* 由於MapReduce排序主要發生在Key2，因此需要將EmpSalaryBean作為Key2的輸出
* 這裏使用mapper類即可，不需要寫Reducer類
* @author songjq
*
*/
static class ObjectSortSingleClumnMapper extends Mapper<LongWritable, Text, EmpSalaryBean1, NullWritable>{
@Override
protected void map(LongWritable key1, Text value1,
Context context)
throws IOException, InterruptedException {
//獲取一行
String line = value1.toString();
//分詞
String[] fds = StringUtils.split(line, ",");
//將分詞數據封裝到EmpSalaryBean對象
EmpSalaryBean1 es = new EmpSalaryBean1();
es.setSeq(new Integer(fds[0]).intValue());
es.setC_oid(fds[1]);
es.setC_employee_name(fds[2]);
es.setC_second_company_name(fds[3]);
es.setC_third_company_name(fds[4]);
es.setC_fourth_company_name(fds[5]);
es.setC_company_name(fds[6]);
es.setC_dept_name(fds[7]);
es.setC_sub_total(new Float(fds[8]).floatValue());
es.setC_com_fund_payamt(new Float(fds[9]).floatValue());
//序列化輸出到Reducer
context.write(es,NullWritable.get());
}
}
/**
* 提交job
* @param args
* @throws Exception
*/
public static void main(String[] args) throws Exception {
Job job = Job.getInstance(new Configuration());
job.setJarByClass(ObjectSortSingleClumn.class);
job.setMapperClass(ObjectSortSingleClumnMapper.class);
job.setMapOutputKeyClass(EmpSalaryBean1.class);
job.setMapOutputValueClass(NullWritable.class);
job.setOutputKeyClass(EmpSalaryBean1.class);
job.setOutputValueClass(NullWritable.class);
FileInputFormat.setInputPaths(job, new Path("D:\\test\\tmp\\sort\\empsalary.csv"));
FileOutputFormat.setOutputPath(job, new Path("D:\\test\\tmp\\sort\\out7"));
job.waitForCompletion(true);
}
}
2018-07-30期 MapReduce對象排序（單列排序）