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

news/2024/12/28 22:28:49/

文章目录

  • Hadoop中MapReduce过程中Shuffle过程实现自定义排序
    • 一、引言
    • 二、实现WritableComparable接口
      • 1、自定义Key类
    • 三、使用Job.setSortComparatorClass方法
      • 2、设置自定义排序器
      • 3、自定义排序器类
    • 四、使用示例
    • 五、总结

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

在这里插入图片描述

一、引言

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

二、实现WritableComparable接口

1、自定义Key类

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

package mr;
import org.apache.hadoop.io.WritableComparable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;public class Employee implements WritableComparable<Employee> {private int empno;private String ename;private String job;private int mgr;private String hiredate;private int sal;private int comm;private int deptno;@Overridepublic String toString(){return "Employee[empno="+empno+",ename="+ename+",sal="+sal+",deptno="+deptno+"]";}@Overridepublic int compareTo(Employee o) {// 多个列的排序:select * from emp order by deptno,sal;// 首先按照deptno排序if(this.deptno > o.getDeptno()){return 1;}else if(this.deptno < o.getDeptno()){return -1;}// 如果deptno相等,按照sal排序if(this.sal >= o.getSal()){return 1;}else{return -1;}}@Overridepublic void write(DataOutput output) throws IOException {// 序列化output.writeInt(this.empno);output.writeUTF(this.ename);output.writeUTF(this.job);output.writeInt(this.mgr);output.writeUTF(this.hiredate);output.writeInt(this.sal);output.writeInt(this.comm);output.writeInt(this.deptno);}@Overridepublic void readFields(DataInput input) throws IOException {// 反序列化this.empno = input.readInt();this.ename = input.readUTF();this.job = input.readUTF();this.mgr = input.readInt();this.hiredate = input.readUTF();this.sal = input.readInt();this.comm = input.readInt();this.deptno = input.readInt();}
}

三、使用Job.setSortComparatorClass方法

2、设置自定义排序器

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

package mr;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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;public class CustomSort {public static class Map extends Mapper<Object, Text, Employee, IntWritable> {private static Employee emp = new Employee();private static IntWritable one = new IntWritable(1);@Overrideprotected void map(Object key, Text value, Context context) throws IOException, InterruptedException {String[] line = value.toString().split("\t");emp.setEmpno(Integer.parseInt(line[0]));emp.setEname(line[1]);emp.setJob(line[2]);emp.setMgr(Integer.parseInt(line[3]));emp.setHiredate(line[4]);emp.setSal(Integer.parseInt(line[5]));emp.setComm(Integer.parseInt(line[6]));emp.setDeptno(Integer.parseInt(line[7]));context.write(emp, one);}}public static class Reduce extends Reducer<Employee, IntWritable, Employee, IntWritable> {@Overrideprotected void reduce(Employee key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {for (IntWritable val : values) {context.write(key, val);}}}public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job job = Job.getInstance(conf, "CustomSort");job.setJarByClass(CustomSort.class);job.setMapperClass(Map.class);job.setReducerClass(Reduce.class);job.setOutputKeyClass(Employee.class);job.setOutputValueClass(IntWritable.class);// 设置自定义排序器job.setSortComparatorClass(EmployeeComparator.class);Path in = new Path("hdfs://localhost:9000/mr/in/customsort");Path out = new Path("hdfs://localhost:9000/mr/out/customsort");FileInputFormat.addInputPath(job, in);FileOutputFormat.setOutputPath(job, out);System.exit(job.waitForCompletion(true) ? 0 : 1);}
}

3、自定义排序器类

package mr;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;public class EmployeeComparator extends WritableComparator {protected EmployeeComparator() {super(Employee.class, true);}@Overridepublic int compare(WritableComparable w1, WritableComparable w2) {Employee e1 = (Employee) w1;Employee e2 = (Employee) w2;// 首先按照deptno排序int deptCompare = Integer.compare(e1.getDeptno(), e2.getDeptno());if (deptCompare != 0) {return deptCompare;}// 如果deptno相等,按照sal排序return Integer.compare(e1.getSal(), e2.getSal());}
}

四、使用示例

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

五、总结

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


版权声明:本博客内容为原创,转载请保留原文链接及作者信息。

参考文章

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

http://www.ppmy.cn/news/1558904.html

相关文章

CASA模型相关遥感数据及MODIS NDVI、FPAR遥感产品数据时序重建

植被作为陆地生态系统的重要组成部分对于生态环境功能的维持具有关键作用。植被净初级生产力&#xff08;Net Primary Productivity, NPP&#xff09;是指单位面积上绿色植被在单位时间内由光合作用生产的有机质总量扣除自养呼吸的剩余部分。植被NPP是表征陆地生态系统功能及可…

uniapp实现APP、小程序与webview页面间通讯

需求&#xff1a; 1、需要在Uniapp开发的APP或小程序页面嵌入一个H5网页&#xff0c;需要拿到H5给APP传递的数据。 2、并且这个H5是使用vuevant开发的。&#xff08;其实跟使用uniapp开发H5一样&#xff09; 实现步骤&#xff1a; 1、首先需要兼容多端和App端&#xff0c;因…

MATLAB语言的网络编程

标题&#xff1a;MATLAB中的网络编程&#xff1a;深入探索与实践 一、引言 在现代科学和工程领域中&#xff0c;网络编程已经成为了数据处理、信号分析、模型构建等众多任务中不可或缺的一环。MATLAB作为一款强大的数学计算软件&#xff0c;不仅提供了丰富的数值计算功能&…

Elasticsearch介绍及安装部署

Elasticsearch介绍 Elasticsearch 是一个分布式搜索引擎&#xff0c;底层基于 Lucene 实现。Elasticsearch 屏蔽了 Lucene 的底层细节&#xff0c;提供了分布式特性&#xff0c;同时对外提供了 Restful API。Elasticsearch 以其易用性迅速赢得了许多用户&#xff0c;被用在网站…

springcloud篇2-feign、gateway

一、Feign(http客户端) 1.1 简介 之前不同的服务之间进行远程调用使用的是RestTemplate。 存在下面的问题&#xff1a; &#xff08;1&#xff09;代码可读性差&#xff0c;编程体验不统一&#xff1b; &#xff08;2&#xff09;参数复杂&#xff0c;URL难以维护。 Feign(…

【从零开始入门unity游戏开发之——C#篇33】C#委托(`Delegate`)和事件(`event` )、事件与委托的区别、Invoke()的解释

文章目录 一、委托&#xff08;Delegate&#xff09;1、什么是委托&#xff1f;2、委托的基本语法3、定义自定义委托4、如何使用自定义委托5、多播委托6、C# 中的系统委托7、GetInvocationList 获取多个函数返回值8、总结 二、事件&#xff08;event &#xff09;1、事件是什么…

柒拾捌- 如何通过数据影响决策(六)- 放大再放大

1、整体带来的错觉 当我们观察宏观的数据时&#xff0c;常常会发现有些东西 无法理解。例如为什么人人都说楼价在跌&#xff0c;但公布的楼价数据却在涨&#xff1f;例如为什么经济感受那么差&#xff0c;宏观数据却还是在涨&#xff1f; 如果我们只在于 某个粒度 的数据&…

农历节日倒计时:基于Python的公历与农历日期转换及节日查询小程序

农历节日倒计时&#xff1a;基于Python的公历与农历日期转换及节日查询小程序 摘要 又是一年春节即将到来&#xff0c;突然想基于Python编写一个农历节日的倒计时小程序。该程序能够根据用户输入的农历节日名称&#xff0c;计算出距离该节日还有多少天。通过使用lunardate库进…