【MapReduce】扩展案例 ---- TopN（求取数据集前N名）_topn maprreduce
定义在map()外面。
·
网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。
一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！
需求
对之前统计的上、下行流量总值输出结果进行加工，输出流量使用量在前10的用户信息
（1）输入数据
（2）输出数据
返回顶部
需求分析
代码实现
① Bean类
- 创建Bean类，封装上下行、总流量对象
- 创建无参构造方法 — 供反序列化使用
- 重写序列化、反序列化方法
- 重写toString()方法
- 继承自WritableComparable，重写compareTo()方法，自定义排序
- 自定义方法统计总流量
package 第七章_MR扩展案例.TopN;
import org.apache.hadoop.io.WritableComparable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
public class TopNBean implements WritableComparable<TopNBean> {
private long upFlow;
private long downFlow;
private long sumFlow;
/\*\*
\* 无参构造方法
\*/
public TopNBean() {
}
/\*\*
\* 重写比较方法
\* @param bean
\* @return
\*/
@Override
public int compareTo(TopNBean bean) {
int result;
if (sumFlow > bean.getSumFlow()) {
result = -1;
} else if (sumFlow < bean.getSumFlow()) {
result = 1;
} else {
result = 0;
}
return result;
}
/\*\*
\* 序列化
\*
\* @param dataOutput
\* @throws IOException
\*/
@Override
public void write(DataOutput dataOutput) throws IOException {
dataOutput.writeLong(upFlow);
dataOutput.writeLong(downFlow);
dataOutput.writeLong(sumFlow);
}
/\*\*
\* 反序列化
\*
\* @param dataInput
\* @throws IOException
\*/
@Override
public void readFields(DataInput dataInput) throws IOException {
upFlow = dataInput.readLong();
downFlow = dataInput.readLong();
sumFlow = dataInput.readLong();
}
@Override
public String toString() {
return upFlow + "\t"+ downFlow + "\t"+ sumFlow;
}
/\*\*
\* set\get
\*
\* @return
\*/
public long getUpFlow() {
return upFlow;
}
public void setUpFlow(long upFlow) {
this.upFlow = upFlow;
}
public long getDownFlow() {
return downFlow;
}
public void setDownFlow(long downFlow) {
this.downFlow = downFlow;
}
public long getSumFlow() {
return sumFlow;
}
public void setSumFlow(long sumFlow) {
this.sumFlow = sumFlow;
}
public void set(long upFlow1, long downFlow1) {
upFlow = upFlow1;
downFlow = downFlow1;
sumFlow = upFlow+downFlow;
}
}
② Mapper 阶段
- 定义TreeMap集合
- 读取数据、拆分、封装Bean对象
- 以Bean对象为key，号码为value键值对存入TreeMap集合，按照自定义排序进行排序
- 存入同时判断集合大小，以达到提取前n名的效果
- 遍历集合写出到Reducer端
package 第七章_MR扩展案例.TopN;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
import java.util.Iterator;
import java.util.TreeMap;
public class TopNMapper extends Mapper<LongWritable, Text, TopNBean, Text> {
// 1.定义一个treeMap作为存储数据的容器（自动按照key排序）
private TreeMap<TopNBean, Text> treeMap = new TreeMap<>();
TopNBean k;
Text v = new Text();
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
// 13470253144 180 180 360
// 13509468723 7335 110349 117684
k = new TopNBean();
// 1.读取一行数据
String line = value.toString();
// 2.拆分
String[] fields = line.split("\t");
// 3.封装Beand对象
long upFlow = Long.parseLong(fields[1]);
long downFlow = Long.parseLong(fields[2]);
long sumFlow = Long.parseLong(fields[3]);
k.setUpFlow(upFlow);
k.setDownFlow(downFlow);
k.setSumFlow(sumFlow);
String phone = fields[0];
v.set(phone);
// 4.向集合中添加元素
treeMap.put(k, v);
// 5.限制TreeMap的数量
if (treeMap.size() > 10) {
treeMap.remove(treeMap.lastKey()); // 删除最后一个
}
}
/\*\*
\* 重写cleanUp，统一释放资源
\*
\* @param context
\* @throws IOException
\* @throws InterruptedException
\*/
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
// 6 遍历treeMap集合，输出数据
Iterator<TopNBean> bean = treeMap.keySet().iterator();
while (bean.hasNext()) {
TopNBean k = bean.next();
context.write(k, treeMap.get(k));
}
}
}
☠ 注意
// 1.定义一个treeMap作为存储数据的容器（自动按照key排序）
private TreeMap<TopNBean, Text> treeMap = new TreeMap<>();
TopNBean k;
Text v = new Text();
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
// 创建Bean对象
k = new TopNBean();
// 1.读取一行数据
// 2.拆分
// 3.封装Beand对象
......
v.set(phone);
// 4.向集合中添加元素
treeMap.put(k, v);
这个地方要重点关注，在创建Bean对象k的时候要写在map()方法内部，它是针对每一次的map(),也就是每一行数据而言的。当你每读取一行数据，就会封装一个Bean对象(需要内存地址不同)，然后写入TreeMap集合；否则定义在map()外面，这样k(Bean对象)就是全局变量，内存地址不会改变，针对每一行数据而言用的就是同一个Bean对象，在封装的时候对原有的Bean对象只是重新赋值，添加到TreeMap集合中的元素也始终只有一个，最终结果只会输出最后一行数据。
③ Reducer 阶段
- 拉取Mapper端数据集，同样创建TreeMapper集合
- 遍历Mapper端数据集存入TreeMap集合，筛选前10条
- 遍历TreeMap集合输出结果
package 第七章_MR扩展案例.TopN;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
import java.util.Iterator;
import java.util.TreeMap;
public class TopNReducer extends Reducer<TopNBean, Text, Text, TopNBean> {
// 定义一个TreeMap作为存储数据的容器（天然按key排序）
TreeMap<TopNBean, Text> flowMap = new TreeMap<>();
@Override
protected void reduce(TopNBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
**既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，涵盖了95%以上大数据知识点，真正体系化！**
**由于文件比较多，这里只是将部分目录截图出来，全套包含大厂面经、学习笔记、源码讲义、实战项目、大纲路线、讲解视频，并且后续会持续更新**
**[需要这份系统化资料的朋友，可以戳这里获取](https://bbs.csdn.net/topics/618545628)**
714991073830)]
[外链图片转存中...(img-8ucRzBsD-1714991073830)]
[外链图片转存中...(img-drtxXz7m-1714991073830)]
**既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，涵盖了95%以上大数据知识点，真正体系化！**
**由于文件比较多，这里只是将部分目录截图出来，全套包含大厂面经、学习笔记、源码讲义、实战项目、大纲路线、讲解视频，并且后续会持续更新**
**[需要这份系统化资料的朋友，可以戳这里获取](https://bbs.csdn.net/topics/618545628)**
魔乐社区（Modelers.cn) 是一个中立、公益的人工智能社区，提供人工智能工具、模型、数据的托管、展示与应用协同服务，为人工智能开发及爱好者搭建开放的学习交流平台。社区通过理事会方式运作，由全产业链共同建设、共同运营、共同享有，推动国产AI生态繁荣发展。
更多推荐
5
0- 0
已为社区贡献1条内容
所有评论(0)