首页 文章 精选 留言 我的

精选列表

搜索[腾讯云双十一抄作业],共10000篇文章
优秀的个人博客,低调大师

hadoop提交作业自定义排序和分组

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/qq1010885678/article/details/43736521 现有数据如下: 3 3 3 2 3 1 2 2 2 1 1 1 要求为: 先按第一列从小到大排序,如果第一列相同,按第二列从小到大排序 如果是hadoop默认的排序方式,只能比较key,也就是第一列,而value是无法参与排序的 这时候就需要用到自定义的排序规则 解决思路: 自定义数据类型,将原本的key和value都包装进去 将这个数据类型当做key,这样就比较key的时候就可以包含第一列和第二列的值了 自定义数据类型NewK2如下: //要实现自定义的排序规则必须实现WritableComparable接口,泛型参数为类本身 public class NewK2 implements WritableComparable<NewK2> { //代表第一列和第二列的数据 Long first; Long second; public NewK2() { } public NewK2(long first, long second) { this.first = first; this.second = second; } //重写序列化和反序列化方法 @Override public void readFields(DataInput in) throws IOException { this.first = in.readLong(); this.second = in.readLong(); } @Override public void write(DataOutput out) throws IOException { out.writeLong(first); out.writeLong(second); } //当k2进行排序时,会自动调用该方法. 当第一列不同时,升序;当第一列相同时,第二列升序 //如果希望降序排列,那么只需要对调this对象和o对象的顺序 @Override public int compareTo(NewK2 o) { if(this.first != o.first) { return (int)(this.first - o.first); } else { return (int) (this.second - o.second); } } //重写hashCode和equals方法 @Override public int hashCode() { return this.first.hashCode() + this.second.hashCode(); } @Override public boolean equals(Object obj) { if (!(obj instanceof NewK2)) { return false; } NewK2 oK2 = (NewK2) obj; return (this.first == oK2.first) && (this.second == oK2.second); } } MyMapper类代码: public class MyMapper extends Mapper<LongWritable, Text, NewK2, LongWritable> { protected void map( LongWritable key, Text value, org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, NewK2, LongWritable>.Context context) throws java.io.IOException, InterruptedException { final String[] splited = value.toString().split("\t"); //分割完成之后的数据如:3,1 分别赋值给k2对象的first和second属性 final NewK2 k2 = new NewK2(Long.parseLong(splited[0]), Long.parseLong(splited[1])); final LongWritable v2 = new LongWritable(Long.parseLong(splited[1])); //将k2当做key输出,这样在排序的时候就会调用NewK2的compareTo方法,里面写的是我们自己的排序规则 context.write(k2, v2); }; } MyReducer类代码: public class MyReducer extends Reducer<NewK2, LongWritable, LongWritable, LongWritable> { protected void reduce( NewK2 k2, java.lang.Iterable<LongWritable> v2s, org.apache.hadoop.mapreduce.Reducer<NewK2, LongWritable, LongWritable, LongWritable>.Context context) throws java.io.IOException, InterruptedException { context.write(new LongWritable(k2.first), new LongWritable( k2.second)); }; } MySubmit类的代码和之前的一样无需改动 运行可得到结果如下图: 如果业务需求又发生了改变,如:上图结果中,第一列相同的,只要列出第二列的值最小的那个选项即可 那么结果应该为 1 1 2 1 3 1 可是我们之前使用的是自定义的数据类型当做key 而hadoop默认的分组策略是所有key相同的选项当做一组 而两个NewK2对象要相等,就必须要first和second属性都相等才行 这时就需要用到自定义的分组策略 自定义分组类如下: //自定义的分组类必须实现RawComparator,泛型参数为类本身 public class MyGroupingComparator implements RawComparator<NewK2> { //重写两个比较方法 //按对象进行比较,规定只要两个NewK2对象的first属性相同就视为相等 @Override public int compare(NewK2 o1, NewK2 o2) { return (int) (o1.first - o2.first); } /** * @param arg0 * 表示第一个参与比较的字节数组 * @param arg1 * 表示第一个参与比较的字节数组的起始位置 * @param arg2 * 表示第一个参与比较的字节数组的偏移量 * * @param arg3 * 表示第二个参与比较的字节数组 * @param arg4 * 表示第二个参与比较的字节数组的起始位置 * @param arg5 * 表示第二个参与比较的字节数组的偏移量 */ @Override public int compare(byte[] arg0, int arg1, int arg2, byte[] arg3, int arg4, int arg5) { return WritableComparator .compareBytes(arg0, arg1, 8, arg3, arg4, 8); } } 在MySubmit代码中加入设置分组策略 // 1.4 TODO 排序、分区 job.setGroupingComparatorClass(MyGroupingComparator.class);再次运行程序可得到如下图的结果:

优秀的个人博客,低调大师

使用老版本的java api提交hadoop作业

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/qq1010885678/article/details/43735491 还是使用之前的单词计数的例子 自定义Mapper类 import java.io.IOException; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapred.MapReduceBase; import org.apache.hadoop.mapred.Mapper; import org.apache.hadoop.mapred.OutputCollector; import org.apache.hadoop.mapred.Reporter; //自定义的Mapper类必须继承MapReduceBase 并且实现Mapper接口 public class JMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, LongWritable> { @Override public void map(LongWritable key, Text value, OutputCollector<Text, LongWritable> collector, Reporter reporter) throws IOException { String[] ss = value.toString().split("\t"); for (String s : ss) { //使用collector.collect而不是context.write collector.collect(new Text(s), new LongWritable(1)); } } } 自定义Reducer类 import java.io.IOException; import java.util.Iterator; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapred.MapReduceBase; import org.apache.hadoop.mapred.OutputCollector; import org.apache.hadoop.mapred.Reducer; import org.apache.hadoop.mapred.Reporter; //自定义的Reducer类必须继承MapReduceBase 并且实现Reducer接口 public class JReducer extends MapReduceBase implements Reducer<Text, LongWritable, Text, LongWritable> { @Override public void reduce(Text key, Iterator<LongWritable> value, OutputCollector<Text, LongWritable> collector, Reporter reporter) throws IOException { long sum = 0; //由于value不在可以用foreach循环,所以用while代替 while (value.hasNext()) { sum += value.next().get(); } collector.collect(key, new LongWritable(sum)); } } 运行提交代码的类JSubmit import java.io.IOException; import java.net.URI; import java.net.URISyntaxException; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapred.FileInputFormat; import org.apache.hadoop.mapred.FileOutputFormat; import org.apache.hadoop.mapred.JobClient; import org.apache.hadoop.mapred.JobConf; import org.apache.hadoop.mapred.TextInputFormat; import org.apache.hadoop.mapred.TextOutputFormat; public class JSubmit { public static void main(String[] args) throws IOException, URISyntaxException, InterruptedException, ClassNotFoundException { Path outPath = new Path("hdfs://localhost:9000/out"); Path inPath = new Path("/home/hadoop/word"); Configuration conf = new Configuration(); FileSystem fs = FileSystem.get(new URI("hdfs://localhost:9000"), conf); if (fs.exists(outPath)) { fs.delete(outPath, true); } // 使用JobConf 而不是Job JobConf job = new JobConf(conf, JSubmit.class); FileInputFormat.setInputPaths(job, inPath); job.setInputFormat(TextInputFormat.class); job.setMapperClass(JMapper.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(LongWritable.class); job.setReducerClass(JReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(LongWritable.class); FileOutputFormat.setOutputPath(job, outPath); job.setOutputFormat(TextOutputFormat.class); // 使用JobClient.runJob而不是job.waitForCompletion JobClient.runJob(job); } } 可以看到 其实老版本的api差别不大,只是用了少数几个类替换了而已 注意老版本api的类虽然和新版本api的类名字很多都是一模一样的 但是所在的包不同,老版本所在的包都是mapred的,而新版本的都在mapreduce

资源下载

更多资源
Spring

Spring

Spring框架(Spring Framework)是由Rod Johnson于2002年提出的开源Java企业级应用框架,旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念,提供核心容器、应用上下文、数据访问集成等模块,支持整合Hibernate、Struts等第三方框架,其适用范围不仅限于服务器端开发,绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux

Rocky Linux(中文名:洛基)是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版,作为CentOS稳定版停止维护后与RHEL(Red Hat Enterprise Linux)完全兼容的开源替代方案,由社区拥有并管理,支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性,采用模块化包装和SELinux安全架构,默认包含GNOME桌面环境及XFS文件系统,支持十年生命周期更新。

Sublime Text

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能,例如代码缩略图,Python的插件,代码段等。还可自定义键绑定,菜单和工具栏。Sublime Text 的主要功能包括:拼写检查,书签,完整的 Python API , Goto 功能,即时项目切换,多选择,多窗口等等。Sublime Text 是一个跨平台的编辑器,同时支持Windows、Linux、Mac OS X等操作系统。

WebStorm

WebStorm

WebStorm 是jetbrains公司旗下一款JavaScript 开发工具。目前已经被广大中国JS开发者誉为“Web前端开发神器”、“最强大的HTML5编辑器”、“最智能的JavaScript IDE”等。与IntelliJ IDEA同源,继承了IntelliJ IDEA强大的JS部分的功能。

用户登录
用户注册