Hadoop_17_MapRduce_案例2_实现用户手机流量统计(ReduceTask并行度控制)
需求:1.统计每一个用户(手机号)所耗费的总上行流量、下行流量,总流量
1.数据如下:保存为.dat文件(因为以\t切分数据,文件格式必须合适)
1363157985066 13726230503 00-FD-07-A4-72-B8:CMCC 120.196.100.82 i02.c.aliimg.com 24 27 2481 24681 200 1363157995052 13826544101 5C-0E-8B-C7-F1-E0:CMCC 120.197.40.4 4 0 264 0 200 1363157991076 13926435656 20-10-7A-28-CC-0A:CMCC 120.196.100.99 2 4 132 1512 200 1363154400022 13926251106 5C-0E-8B-8B-B1-50:CMCC 120.197.40.4 4 0 240 0 200 1363157993044 18211575961 94-71-AC-CD-E6-18:CMCC-EASY 120.196.100.99 iface.qiyi.com 视频网站 15 12 1527 2106 200 1363157995074 84138413 5C-0E-8B-8C-E8-20:7DaysInn 120.197.40.4 122.72.52.12 20 16 4116 1432 200 1363157993055 13560439658 C4-17-FE-BA-DE-D9:CMCC 120.196.100.99 18 15 1116 954 200 1363157995033 15920133257 5C-0E-8B-C7-BA-20:CMCC 120.197.40.4 sug.so.360.cn 信息安全 20 20 3156 2936 200 1363157983019 13719199419 68-A1-B7-03-07-B1:CMCC-EASY 120.196.100.82 4 0 240 0 200 1363157984041 13660577991 5C-0E-8B-92-5C-20:CMCC-EASY 120.197.40.4 s19.cnzz.com 站点统计 24 9 6960 690 200 1363157973098 15013685858 5C-0E-8B-C7-F7-90:CMCC 120.197.40.4 rank.ie.sogou.com 搜索引擎 28 27 3659 3538 200 1363157986029 15989002119 E8-99-C4-4E-93-E0:CMCC-EASY 120.196.100.99 www.umeng.com 站点统计 3 3 1938 180 200 1363157992093 13560439658 C4-17-FE-BA-DE-D9:CMCC 120.196.100.99 15 9 918 4938 200 1363157986041 13480253104 5C-0E-8B-C7-FC-80:CMCC-EASY 120.197.40.4 3 3 180 180 200 1363157984040 13602846565 5C-0E-8B-8B-B6-00:CMCC 120.197.40.4 2052.flash2-http.qq.com 综合门户 15 12 1938 2910 200 1363157995093 13922314466 00-FD-07-A2-EC-BA:CMCC 120.196.100.82 img.qfc.cn 12 12 3008 3720 200 1363157982040 13502468823 5C-0A-5B-6A-0B-D4:CMCC-EASY 120.196.100.99 y0.ifengimg.com 综合门户 57 102 7335 110349 200 1363157986072 18320173382 84-25-DB-4F-10-1A:CMCC-EASY 120.196.100.99 input.shouji.sogou.com 搜索引擎 21 18 9531 2412 200 1363157990043 13925057413 00-1F-64-E1-E6-9A:CMCC 120.196.100.55 t3.baidu.com 搜索引擎 69 63 11058 48243 200 1363157988072 13760778710 00-FD-07-A4-7B-08:CMCC 120.196.100.82 2 2 120 120 200 1363157985066 13726238888 00-FD-07-A4-72-B8:CMCC 120.196.100.82 i02.c.aliimg.com 24 27 2481 24681 200 1363157993055 13560436666 C4-17-FE-BA-DE-D9:CMCC 120.196.100.99 18 15 1116 954 200
2.技术实现过程:
1.首先将Map输入中的手机号,上行流量,下行流量数据抽取出来(每一行输入数据调用一次自定义map方法处理数据),
然后根据相同的key进行数据分发,以便于相同key会到同一个ReduceTask
2.Map输出为<手机号,bean>,自定义javaBean来封装流量信息,并将javaBean充当Map输出的Value来传输,javaBean
要实现Writable序列化接口,实现两个方法
3.Reduce在获得<手机号,list>后进行累积,然后输出结果即可(框架每传递进来一个kv组,reduce方法被调用一次)
3.代码:FlowCount.java
package cn.bigdata.hdfs.flowsum; import java.io.IOException; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;public class FlowCount {static class FlowCountMapper extends Mapper<LongWritable, Text, Text, FlowBean>{@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//将一行内容转成stringString line = value.toString();//切分字段String[] fields = line.split("\t");//取出手机号String phoneNbr = fields[1];//取出上行流量下行流量long upFlow = Long.parseLong(fields[fields.length-3]);long dFlow = Long.parseLong(fields[fields.length-2]);context.write(new Text(phoneNbr), new FlowBean(upFlow, dFlow));}}static class FlowCountReducer extends Reducer<Text, FlowBean, Text, FlowBean>{//<183323,bean1><183323,bean2><183323,bean3><183323,bean4>....... @Overrideprotected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {long sum_upFlow = 0;long sum_dFlow = 0;//遍历所有bean,将其中的上行流量,下行流量分别累加for(FlowBean bean: values){sum_upFlow += bean.getUpFlow();sum_dFlow += bean.getdFlow();}FlowBean resultBean = new FlowBean(sum_upFlow, sum_dFlow);context.write(key, resultBean);}}public static void main(String[] args) throws Exception {Configuration conf = new Configuration();/*conf.set("mapreduce.framework.name", "yarn");conf.set("yarn.resoucemanager.hostname", "mini1");*/Job job = Job.getInstance(conf);/*job.setJar("/home/hadoop/wc.jar");*///指定本程序的jar包所在的本地路径job.setJarByClass(FlowCount.class);//指定本业务job要使用的mapper/Reducer业务类job.setMapperClass(FlowCountMapper.class);job.setReducerClass(FlowCountReducer.class);//指定mapper输出数据的kv类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(FlowBean.class);//指定最终输出的数据的kv类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(FlowBean.class);//指定job的输入原始文件所在目录FileInputFormat.setInputPaths(job, new Path(args[0]));//指定job的输出结果所在目录FileOutputFormat.setOutputPath(job, new Path(args[1]));//将job中配置的相关参数,以及job所用的java类所在的jar包,提交给yarn去运行/*job.submit();*/boolean res = job.waitForCompletion(true);System.exit(res?0:1);} }
FlowBean.java
如果想在Reducer的输出结果中使用自定义的数据类型,重写FlowBean的toString()方法即可。
package cn.bigdata.hdfs.flowsum; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; import org.apache.hadoop.io.Writable; public class FlowBean implements Writable{ private long upFlow;private long dFlow;private long sumFlow;//反序列化时,需要反射调用空参构造函数,所以要显示定义一个public FlowBean(){}public FlowBean(long upFlow, long dFlow) {this.upFlow = upFlow;this.dFlow = dFlow;this.sumFlow = upFlow + dFlow;}public long getUpFlow() {return upFlow;}public void setUpFlow(long upFlow) {this.upFlow = upFlow;}public long getdFlow() {return dFlow;}public void setdFlow(long dFlow) {this.dFlow = dFlow;}public long getSumFlow() {return sumFlow;}public void setSumFlow(long sumFlow) {this.sumFlow = sumFlow;}/*** 序列化方法*/@Overridepublic void write(DataOutput out) throws IOException {out.writeLong(upFlow);out.writeLong(dFlow);out.writeLong(sumFlow);}/*** 反序列化方法* 注意:反序列化的顺序跟序列化的顺序完全一致*/@Overridepublic void readFields(DataInput in) throws IOException {upFlow = in.readLong();dFlow = in.readLong();sumFlow = in.readLong();}@Overridepublic String toString() { return upFlow + "\t" + dFlow + "\t" + sumFlow;} }
4.执行程序:
4.1.创建HDFS文件存放目录:hadoop fs -mkdir -p /wordcount/phoneFlum
4.2.运行MapReduce程序jar包:
hadoop jar flowsum.jar cn.bigdata.hdfs.flowsum.FlowCount /wordcount/phoneFlum /wordcount/phoneFlumOut
5.查看执行结果:
需求:2.将流量统计结果按照手机归属地省份不同输出到不同文件中(ReduceTask并行度控制,自定义Partitioner)
2.技术实现过程:
1.Mapreduce中会将map输出的kv对,按照相同key分组(调用getPartition),然后分发给不同的reducetask
2.Map输出结果的时候调用了Partitioner组件(返回分区号),由它决定将数据放到哪个区中,默认的分组规则为
:根据key的hashcode%reducetask数来分发,源代码如下:
public class HashPartitioner<K, V> extends Partitioner<K, V> {/** Use {@link Object#hashCode()} to partition. */public int getPartition(K key, V value,int numReduceTasks) {return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;} }
3.所以:如果要按照我们自己的需求进行分组,则需要改写数据分发(分组)组件Partitioner,自定义一个
CustomPartitioner继承抽象类:Partitioner,来返回一个分区编号
4.然后在job对象中,设置自定义partitioner: job.setPartitionerClass(CustomPartitioner.class)
5.自定义partition后,要根据自定义partitioner的逻辑设置相应数量的ReduceTask
3.代码实现自定义partitioner数据分区规则:
package cn.bigdata.hdfs.flowsum; import java.util.HashMap; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Partitioner; /*** Partitioner<Text, FlowBean>中分别 对应的是map输出kv的类型*/ public class ProvincePartitioner extends Partitioner<Text, FlowBean>{public static HashMap<String, Integer> proviceDict = new HashMap<String, Integer>();static{//分为5个区proviceDict.put("136", 0);proviceDict.put("137", 1);proviceDict.put("138", 2);proviceDict.put("139", 3);}@Overridepublic int getPartition(Text key, FlowBean value, int numPartitions) {String prefix = key.toString().substring(0, 3);Integer provinceId = proviceDict.get(prefix);return provinceId==null?4:provinceId;} }
//指定我们自定义的数据分区器 job.setPartitionerClass(ProvincePartitioner.class); //同时指定相应“分区”数量的reducetask job.setNumReduceTasks(5);
运行程序:hadoop jar flowsum.jar cn.bigdata.hdfs.flowsum.FlowCount /wordcount/phoneFlum /wordcount/phoneFlumOut1
此时生成了五个分区文件:
注意:如果reduceTask的数量>= getPartition的结果数 ,则会多产生几个空的输出文件part-r-000xx
如果1<reduceTask的数量<getPartition的结果数 ,则有一部分分区数据无处安放,会Exception
如果 reduceTask的数量=1,则不管mapTask端输出多少个分区文件,最终结果都交给这一个reduceTask,
最终也就只会产生一个结果文件 part-r-00000
需求:3.将统计结果按照总流量倒序排序
//指定我们自定义的数据分区器
job.setPartitionerClass(ProvincePartitioner.class);
//同时指定相应“分区”数量的reducetask
job.setNumReduceTasks(5);
转载于:https://www.cnblogs.com/yaboya/p/9204932.html
Hadoop_17_MapRduce_案例2_实现用户手机流量统计(ReduceTask并行度控制)相关推荐
- 实现用户手机流量统计(ReduceTask并行度控制)
需求:1.实现用户手机流量统计(ReduceTask并行度控制) 数据如下:保存为.dat文件(因为以\t切分数据,文件格式必须合适) 13726230503 00-FD-07-A4-72-B8:CM ...
- android统计app流量的软件,流量控(手机流量统计)app
流量控(手机流量统计)app,为你带来准确,及时的流量通知,帮助用户时刻查询自己的手机流量和使用状况,管理程序的流量使用,全方面保证您的流量使用 流量控(手机流量统计)app介绍 流量控是一款统计安卓 ...
- Hadoop实战学习(3)-手机流量统计
先装备一个txt文档,内容如下: 13726230501 200 1100 13396230502 300 1200 13892730503 400 1300 13897230503 100 300 ...
- MapReduce案例:手机流量的统计
案例:手机流量的统计 对于记录用户手机信息的文件,得出统计每一个用户(手机号)所耗费的总上行流量.下行流量,总流量结果. 分析 1. 实现自定义的 bean 来封装流量信息,使用手机号码作为Key,B ...
- spark企业经典案例之手机app流量统计
spark企业经典案例之手机app流量统计,本课程是基于企业真实项目案例中的一个模块为背景讲解,此业务涉及spark统计,然后通过phpweb进行读取数据,最终通过手机端展示,根据公司案例抽取出来的模 ...
- 手机上网流量统计_数据统计 | 上半年手机流量同比增110.2%,你贡献了多少?
来源:工信部网站.中新经纬 版权申明:内容来源网络,版权归原创者所有.除非无法确认,我们都会标明作者及出处,如有侵权烦请告知我们,我们会立即删除并表示歉意.谢谢! 7月25日,工信部网站公布了2019 ...
- Axure电商后台业务管理系统原型模板/通用版电商后台管理系统/订单管理/营销管理/运营管理/财务管理/统计分析/库存管理/流量统计/运营管理/用户管理/秒杀促销/交易统计/活动管理/广告管理
Axure电商后台业务管理系统原型模板/通用版电商后台管理系统/订单管理/营销管理/运营管理/财务管理/统计分析/库存管理/流量统计/运营管理/用户管理/秒杀促销/交易统计/活动管理/广告管理 作品名 ...
- 手机病毒是盗取手机流量的真凶 用户应时刻警惕
"流量偷跑"成为近期通信行业的新热点.从"我是马甲马老师"的出国流量偷跑乌龙事件,到近期有用户3小时用掉23GB流量,再到某用户1天跑1GB流量的新闻纷纷登上各 ...
- 手机上网流量统计_上半年手机上网流量超552亿GB,你每个月贡献多少?????
7月25日消息,工信部近日公布了<2019年上半年通信业经济运行情况>.数据显示,上半年,移动互联网累计流量达554亿GB,同比增长107.3%:其中通过手机上网的流量达到552亿GB,占 ...
- 数据分析案例——流量统计
使用java处理日志文件,完成需求. 简介 http.log日志文件中,是电信运营商记录用户上网访问某些网站行为的日志记录数据,一条数据中有多个字段用空格或制表符分隔. 例如:"186111 ...
最新文章
- WINCE5.0和WINCE6.0的内存与系统架构
- linux/centos/集群安装ntp时间同步,自身实验成功
- 在有序但含有空的数组中查找字符串
- 拉格朗日乘数法(一种寻找变量受一个或多个条件所限制的多元函数的极值的方法)
- js map 只输出key_什么时候适合使用Map而不是Object
- 翻译:集群索引:通往SQL Server索引级别3的阶梯
- javadoc时候乱码-编码 GBK 的不可映射字符 - wqjsir的专栏 - 博客频道 - CSDN.NET
- python优化网站_[练习] 用PYTHON来优化网站中的图片
- Andrew Ng Machine Learning 专题【Logistic Regression amp; Regularization】
- Java把科学计数法转换为字符串
- Android开发自定义View之滑动按钮与自定义属性
- 跨时代比较:工业化因素是关键
- 16进制发送 mqtt客户端调试工具_MQTT Simulate Device(MQTT客户端调试工具)下载 v1.0.7免费版-下载啦...
- 创意火焰燃烧效果PS动作
- mkv无损转换成mp4,方法步骤
- 学Python用什么系统?
- [计算机组成原理,Cache总容量的计算方法]假设主存地址位数是32位,按字节编址,贮存和Cache采用全相联映射的方式,主存块大小为1个字,每字32位,采用回写法和随机替换的策略,则能存放32K字数
- CorelDRAW 2020Mac、2021mac 版本正式回归 macOS免激活版
- Azure Messaging
- CodeForces - 985E Pencils and Boxes
热门文章
- python爬虫学习研究
- okHttp3自用封装
- How to recover deleted data from SQL Server
- java调用perl脚本_Java中调用Perl脚本的应用
- Netlink 0002 -- 什么是Netlink
- Linux内核多线程——补充(各种平台下的多线程)
- linux调度器(十)——调度器/proc信息解读
- SylixOS allwinner h6 链接脚本
- 1079 Total Sales of Supply Chain(25 分)
- mysqld和mysql区别_mysqld与mysqld_safe的区别