Hadoop中MapReduce过程中Shuffle过程实现自定义排序

Hadoop中MapReduce过程中Shuffle过程实现自定义排序

一、引言

MapReduce框架中的Shuffle过程是连接Map阶段和Reduce阶段的桥梁，负责将Map任务的输出结果按照key进行分组和排序，并将雷同key的数据传递给对应的Reduce任务进行处理。Shuffle过程的性能直接影响到整个MapReduce作业的实行效率。在默认情况下，Hadoop使用TotalOrderPartitioner进行排序，但偶然我们需要根据特定的业务逻辑进行自定义排序。本文将介绍两种方法来实现自定义排序：实现WritableComparable接口和使用Job.setSortComparatorClass方法。下面是详细的步骤和代码示例。
二、实现WritableComparable接口

1、自定义Key类

起首，我们需要定义一个类并实现WritableComparable接口，该接口要求实现compareTo方法，用于定义排序逻辑。

1. package mr;
2. import org.apache.hadoop.io.WritableComparable;
3. import java.io.DataInput;
4. import java.io.DataOutput;
5. import java.io.IOException;
7. public class Employee implements WritableComparable<Employee> {
8.     private int empno;
9.     private String ename;
10.     private String job;
11.     private int mgr;
12.     private String hiredate;
13.     private int sal;
14.     private int comm;
15.     private int deptno;
17.     @Override
18.     public String toString(){
19.         return "Employee[empno="+empno+",ename="+ename+",sal="+sal+",deptno="+deptno+"]";
20.     }
22.     @Override
23.     public int compareTo(Employee o) {
24.         // 多个列的排序：select * from emp order by deptno,sal;
25.         // 首先按照deptno排序
26.         if(this.deptno > o.getDeptno()){
27.             return 1;
28.         }else if(this.deptno < o.getDeptno()){
29.             return -1;
30.         }
31.         // 如果deptno相等，按照sal排序
32.         if(this.sal >= o.getSal()){
33.             return 1;
34.         }else{
35.             return -1;
36.         }
37.     }
39.     @Override
40.     public void write(DataOutput output) throws IOException {
41.         // 序列化
42.         output.writeInt(this.empno);
43.         output.writeUTF(this.ename);
44.         output.writeUTF(this.job);
45.         output.writeInt(this.mgr);
46.         output.writeUTF(this.hiredate);
47.         output.writeInt(this.sal);
48.         output.writeInt(this.comm);
49.         output.writeInt(this.deptno);
50.     }
52.     @Override
53.     public void readFields(DataInput input) throws IOException {
54.         // 反序列化
55.         this.empno = input.readInt();
56.         this.ename = input.readUTF();
57.         this.job = input.readUTF();
58.         this.mgr = input.readInt();
59.         this.hiredate = input.readUTF();
60.         this.sal = input.readInt();
61.         this.comm = input.readInt();
62.         this.deptno = input.readInt();
63.     }
64. }

复制代码

三、使用Job.setSortComparatorClass方法

2、设置自定义排序器

除了实现WritableComparable接口外，我们还可以使用Job.setSortComparatorClass方法来设置自定义排序器。这种方法答应我们在不修改Key类的情况下实现自定义排序。

1. package mr;
2. import org.apache.hadoop.conf.Configuration;
3. import org.apache.hadoop.fs.Path;
4. import org.apache.hadoop.io.NullWritable;
5. import org.apache.hadoop.mapreduce.Job;
6. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
7. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
9. public class CustomSort {
10.     public static class Map extends Mapper<Object, Text, Employee, IntWritable> {
11.         private static Employee emp = new Employee();
12.         private static IntWritable one = new IntWritable(1);
14.         @Override
15.         protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
16.             String[] line = value.toString().split("\t");
17.             emp.setEmpno(Integer.parseInt(line[0]));
18.             emp.setEname(line[1]);
19.             emp.setJob(line[2]);
20.             emp.setMgr(Integer.parseInt(line[3]));
21.             emp.setHiredate(line[4]);
22.             emp.setSal(Integer.parseInt(line[5]));
23.             emp.setComm(Integer.parseInt(line[6]));
24.             emp.setDeptno(Integer.parseInt(line[7]));
25.             context.write(emp, one);
26.         }
27.     }
29.     public static class Reduce extends Reducer<Employee, IntWritable, Employee, IntWritable> {
30.         @Override
31.         protected void reduce(Employee key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
32.             for (IntWritable val : values) {
33.                 context.write(key, val);
34.             }
35.         }
36.     }
38.     public static void main(String[] args) throws Exception {
39.         Configuration conf = new Configuration();
40.         Job job = Job.getInstance(conf, "CustomSort");
41.         job.setJarByClass(CustomSort.class);
42.         job.setMapperClass(Map.class);
43.         job.setReducerClass(Reduce.class);
44.         job.setOutputKeyClass(Employee.class);
45.         job.setOutputValueClass(IntWritable.class);
46.         // 设置自定义排序器
47.         job.setSortComparatorClass(EmployeeComparator.class);
48.         
49.         Path in = new Path("hdfs://localhost:9000/mr/in/customsort");
50.         Path out = new Path("hdfs://localhost:9000/mr/out/customsort");
51.         FileInputFormat.addInputPath(job, in);
52.         FileOutputFormat.setOutputPath(job, out);
54.         System.exit(job.waitForCompletion(true) ? 0 : 1);
55.     }
56. }

复制代码

3、自定义排序器类

1. package mr;
2. import org.apache.hadoop.io.WritableComparable;
3. import org.apache.hadoop.io.WritableComparator;
5. public class EmployeeComparator extends WritableComparator {
6.     protected EmployeeComparator() {
7.         super(Employee.class, true);
8.     }
10.     @Override
11.     public int compare(WritableComparable w1, WritableComparable w2) {
12.         Employee e1 = (Employee) w1;
13.         Employee e2 = (Employee) w2;
14.         // 首先按照deptno排序
15.         int deptCompare = Integer.compare(e1.getDeptno(), e2.getDeptno());
16.         if (deptCompare != 0) {
17.             return deptCompare;
18.         }
19.         // 如果deptno相等，按照sal排序
20.         return Integer.compare(e1.getSal(), e2.getSal());
21.     }
22. }

复制代码

四、使用示例

下面是一个简单的MapReduce示例，展示了Shuffle过程在实际应用中的使用。这个示例中，我们使用了自定义的Employee类作为Key，并设置了自定义的排序器EmployeeComparator。
五、总结

通过实现WritableComparable接口和使用Job.setSortComparatorClass方法，我们可以在Hadoop MapReduce过程中实现自定义排序。这两种方法提供了灵活的排序机制，答应我们根据不同的业务需求对数据进行排序处理，从而提高数据处理的效率和正确性。

---

版权声明：本博客内容为原创，转载请保留原文链接及作者信息。
参考文章：

• Hadoop之mapreduce数据排序案例（详细代码）
  
• Java Job.setSortComparatorClass方法代码示例
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。