阿里买域名 电脑做网站,做简历哪个网站好,网站 虚拟空间,网站建设vs网络推广hadoop学习之MapReduce案例#xff1a;输出每个班级中的成绩前三名的学生
所要处理的数据案例#xff1a;
1500100001 施笑槐,22,女,文科六班,406
1500100002 吕金鹏,24,男,文科六班,440
1500100003 单乐蕊,22,女,理科六班,359
1500100004 葛德曜,24,男,理科三班,421
15001…hadoop学习之MapReduce案例输出每个班级中的成绩前三名的学生
所要处理的数据案例
1500100001 施笑槐,22,女,文科六班,406
1500100002 吕金鹏,24,男,文科六班,440
1500100003 单乐蕊,22,女,理科六班,359
1500100004 葛德曜,24,男,理科三班,421
1500100005 宣谷芹,22,女,理科五班,395
1500100006 边昂雄,21,男,理科二班,314
...1.Map端
package com.shujia.mr.top3;import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;/*TODO在编写代码之前需要先定义数据的处理逻辑对于各班级中的学生总分进行排序要求取出各班级中总分前三名学生MapTask阶段① 读取ReduceJoin的处理结果并对数据进行提取② 按照学生的班级信息对班级作为Key,整行数据作为Value写出到 ReduceTask 端ReduceTask阶段① 接收到整个班级中的所有学生信息并将该数据存放在迭代器中*/public class Top3Mapper extends MapperLongWritable, Text, Text, Stu {/*** 直接将学生对象发送到Reduce端进行操作* ① 对于Stu学生自定义学生类作为输出类型需要将当前类进行序列化操作 implement Writable 接口* ② 同时需要在自定义类中保证 类是具有无参构造的* 运行时会出现* java.lang.RuntimeException: java.lang.NoSuchMethodException: com.shujia.mr.top3.Stu.init()* 从日志上可以看到调用了 Stu.init() 指定的就是无参构造* 从逻辑上* 在Mapper端 构建了Stu对象 通过调用其 write 对其进行了序列化操作* 在Reducer端 需要对其进行反序列化 通过无参构造创建自身的空参对象 调用readFields方法进行 反序列化* 将数据赋予给当前的空参对象属性*/Overrideprotected void map(LongWritable key, Text value, MapperLongWritable, Text, Text, Stu.Context context) throws IOException, InterruptedException {// 1500100009 沈德昌,21,男,理科一班,251 表示读取到的数据String[] split value.toString().split(\t);if (split.length 2) {String otherInfo split[1];String[] columns otherInfo.split(,);if (columns.length 5) {String clazz columns[3];Stu stu new Stu(split[0], columns[0], Integer.valueOf(columns[1]), columns[2], columns[3], Integer.valueOf(columns[4]));context.write(new Text(clazz), stu);}}}
}2.Reduce端
package com.shujia.mr.top3;import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;/*TODO ReduceTask阶段*/
public class Top3Reducer extends ReducerText, Stu, Text, NullWritable {/*** 对一个班级中所有的学生成绩进行排序 * 1.将数据存储在一个容器中* 2.对容器中数据进行排序操作* 对排序的结果进行取前三** param key 表示班级信息* param values 一个班级中所有的学生对象* param context* throws IOException* throws InterruptedException*/Overrideprotected void reduce(Text key, IterableStu values, ReducerText, Stu, Text, NullWritable.Context context) throws IOException, InterruptedException {/*TODO 当程序执行到Reducer端时需要对Values中的数据进行遍历获取每一个学生对象但是在添加过程中ArrayList中所有的对象信息都变成一样的。表示当前 ArrayList存储的对象为1个每次添加的引用信息都是指向一个对象地址如何解决每次获取到对象后对其进行克隆一份重新创建一个对象进行存储*/ArrayListStu stus new ArrayList();for (Stu stu : values) {// 排序方案2需要对Stu进行序列化//TODO 每次获取到对象后对其进行克隆一份重新创建一个对象进行存储Stu stu1 new Stu(stu.id, stu.name, stu.age, stu.gender, stu.clazz, stu.score);stus.add(stu1);}// 进行排序操作将stus集合传入函数进行排序Collections.sort(stus,// 设定排序规则new ComparatorStu() {/*CSDNreturn 0:不交换位置不排序return 1:交换位置return -1:不交换位置return o1-o2:升序排列return o2-o1:降序排列*/Overridepublic int compare(Stu o1, Stu o2) {int compareScore o1.score - o2.score;// 保证成绩序列是降序排序若成绩相同则按照学号进行字典排序返回数值最后进行字典 升序排序//String中的compareTo方法用字符串1跟字符串2作比较如果字符串1的字典顺序在字符串2前面较小则返回一个负数。// 若在后面则返回一个正数。若两个字符串的字典顺序相同则返回0。return -compareScore 0 ? 1 : (compareScore 0 ? o1.id.compareTo(o2.id) : -1);}});// 对排序的结果进行遍历for (int i 0; i 3; i) {context.write(new Text(stus.get(i).toString(),(i1)),NullWritable.get());}}
}3.main方法
package com.shujia.mr.top3;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;import java.io.FileNotFoundException;
import java.io.IOException;public class Top3 {/*TODO:将项目打包到Hadoop中进行执行。*/public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {// TODO MapReduce程序入口中的固定写法// TODO 1.获取Job对象 并设置相关Job任务的名称及入口类Configuration conf new Configuration();Job job Job.getInstance(conf, Top3);// 设置当前main方法所在的入口类job.setJarByClass(Top3.class);// TODO 2.设置自定义的Mapper和Reducer类job.setMapperClass(Top3Mapper.class);job.setReducerClass(Top3Reducer.class);// TODO 3.设置Mapper的KeyValue输出类 和 Reducer的输出类 (最终输出)job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(Stu.class);job.setOutputKeyClass(Text.class);/*NullWritable是Writable的一个特殊类实现方法为空实现不从数据流中读数据也不写入数据只充当占位符如在MapReduce中如果你不需要使用键或值你就可以将键或值声明为NullWritable,NullWritable是一个不可变的单实例类型。*/job.setOutputValueClass(NullWritable.class);// TODO 4.设置数据的输入和输出路径// 本地路径FileSystem fileSystem FileSystem.get(job.getConfiguration());Path outPath new Path(hadoop/out/new_top3);
// Path outPath new Path(/data/hadoop/out/new_top3);Path inpath new Path(hadoop/out/reducejoin);
// Path inpath new Path(/data/hadoop/out/reducejoin);if (!fileSystem.exists(inpath)) {throw new FileNotFoundException(inpath不存在);}TextInputFormat.addInputPath(job,inpath);if (fileSystem.exists(outPath)) {System.out.println(路径存在开始删除);fileSystem.delete(outPath,true);}TextOutputFormat.setOutputPath(job,outPath);// TODO 5.提交任务开始执行job.waitForCompletion(true);}
}4.创建的学生类
package com.shujia.mr.top3;import org.apache.hadoop.io.Writable;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.io.Serializable;public class Stu implements Writable {String id;String name;int age;String gender;String clazz;int score;/*TODO 使用Hadoop序列化的问题java.lang.RuntimeException: java.lang.NoSuchMethodException: com.shujia.mr.top3.Stu.init()*///TODO 需要给定无参构造方法序列化需要public Stu() {}public Stu(String id, String name, int age, String gender, String clazz, int score) {this.id id;this.name name;this.age age;this.gender gender;this.clazz clazz;this.score score;}Overridepublic String toString() {return id , name , age , gender , clazz , score;}// TODO 自定义类要重写下列方法才能进行序列化/*对于Write方法中是对当前的对象进行序列化操作*/Overridepublic void write(DataOutput out) throws IOException {out.writeUTF(id);out.writeUTF(name);out.writeInt(age);out.writeUTF(gender);out.writeUTF(clazz);out.writeInt(score);}/*readFields方法中是对当前对象进行反序列化操作*/Overridepublic void readFields(DataInput in) throws IOException {this.id in.readUTF(); // 将0101数据反序列化数据并保存到当前属性中this.name in.readUTF();this.age in.readInt();this.gender in.readUTF();this.clazz in.readUTF();this.score in.readInt();}
}
