使用mapreduce进行流量汇总的教程
程序员文章站
2022-07-03 20:58:55
现有文件关于流量文件内容如下
1363157985066 13726230503 00-FD-07-A4-72-B8:CMCC 120.196.100.82 i02.c.alii...
现有文件关于流量文件内容如下
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 ...
其中第二列表示手机号，倒数第三列表示上下流量，倒数第二列表示下行流量。
要求：统计每个用户（手机号）所耗费的上、下行流量以及总流量。
分析：
map()：取出一行信息，切分，取出手机号，上下行流量（封装到一个bean中）。写出context.write(手机号，bean)。
reduce()：拿到一个号码对应的value（迭代器），分别取出bean计算上下行以及总流量。最终封装到bean中，写出context.write(手机号，bean)。
程序实现
为了方便就写在了一个类里面
public class Flowcount {
/**
* KEYIN：默认情况下，是mr框架所读到的一行文本的起始偏移量，Long，但是在hadoop中有自己的
* 更精简的序列化接口(Seria会将类结构都序列化，而实际我们只需要序列化数据)，所以不直接用Long，而用LongWritable
* VALUEIN：默认情况下，是mr框架所读到的一行文本的内容，String,同上，用Text
* KEYOUT：是用户自定义逻辑处理完成之后输出数据中的key
* VALUEOUT：是用户自定义逻辑处理完成之后输出数据中的value
* @author 12706
*
*/
static class FlowcountMapper extends Mapper{
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
//输入为1234 23455 33333 33333(中间是制表符)
//第二列为手机号，倒数第二列为下行流量，倒数第三列为上行流量
String line = value.toString();
String[] values = line.split("\t");
//获取手机号
String phoneNum = values[1];
//获取上行流量下行流量
long upFlow = new Long(values[values.length-3]);
long downFlow = new Long(values[values.length-2]);
//封装好后写出到输出收集器
context.write(new Text(phoneNum), new FlowBean(upFlow,downFlow));
}
}
/**
* KEYIN VALUEIN对应mapper输出的KEYOUT KEYOUT类型对应
* KEYOUT,VALUEOUT：是自定义reduce逻辑处理结果的输出数据类型
* KEYOUT
* VALUEOUT
* @author 12706
*
*/
static class FlowcountReducer extends Reducer{
@Override
protected void reduce(Text key, Iterable beans,Context context)
throws IOException, InterruptedException {
//传进来的实例<13345677654,beans>,即多个该电话的键值对
//取出values获得上下行和总流量求和
long upFlow = 0;
long downFlow = 0;
for (FlowBean flowBean : beans) {
upFlow += flowBean.getUpFlow();
downFlow += flowBean.getDownFlow();
}
context.write(key, new FlowBean(upFlow,downFlow));
}
}
/**
* 相当于一个yarn集群的客户端
* 需要在此封装mr程序的相关运行参数，指定jar包
* 最后提交给yarn
* @author 12706
* @param args
* @throws Exception
*
*/
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
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去运行
boolean b = job.waitForCompletion(true);
System.exit(b?0:1);
}
}
public class FlowBean implements Writable{
private long upFlow;//上行流量
private long downFlow;//下行流量
private long totalFlow;//总流量
//序列化时需要无参构造方法
public FlowBean() {
}
public FlowBean(long upFlow, long downFlow) {
this.upFlow = upFlow;
this.downFlow = downFlow;
this.totalFlow = upFlow + downFlow;
}
//序列化方法 hadoop的序列化很简单，要传递的数据写出去即可
public void write(DataOutput out) throws IOException {
out.writeLong(upFlow);
out.writeLong(downFlow);
out.writeLong(totalFlow);
}
//反序列化方法 注意：反序列化的顺序跟序列化的顺序完全一致
public void readFields(DataInput in) throws IOException {
this.upFlow = in.readLong();
this.downFlow = in.readLong();
this.totalFlow = in.readLong();
}
//重写toString以便展示
@Override
public String toString() {
return upFlow + "\t" + downFlow + "\t" + totalFlow;
}
get，set方法
}
流量汇总测试：
将这个工程打成jar包flowcount.jar(右击工程选择export再选jar file选项)放在本地，将打好的jar包上传到linux，将上面需要汇总的文件也也传到linux（名字叫flow.data）,创建输入文件目录将文件上传。
创建目录 [root@mini2 ~]# hadoop fs -mkdir -p /flowcount/input 需要汇总的文件上传 [root@mini2 ~]# hadoop fs -put -flow.data /flowcount/input 运行 [root@mini2 ~]# hadoop jar flowcount.jar com.scu.hadoop.mr.Flowcount /flowcount/input /flowcount/output
查看目录结构和输出文件汇总结果
[root@mini2 ~]# hadoop fs -ls /flowcount/output/ -rw-r--r-- 2 root supergroup 0 2017-10-09 07:11 /flowcount/output/_SUCCESS -rw-r--r-- 2 root supergroup 551 2017-10-09 07:11 /flowcount/output/part-r-00000 [root@mini2 ~]# hadoop fs -cat /flowcount/output/part-r-00000 13480253104 180 180 360 13502468823 7335 110349 117684 13560436666 1116 954 2070 13560439658 2034 5892 7926 13602846565 1938 2910 4848 13660577991 6960 690 7650 13719199419 240 0 240 13726230503 2481 24681 27162 13726238888 2481 24681 27162 13760778710 120 120 240 13826544101 264 0 264 13922314466 3008 3720 6728 13925057413 11058 48243 59301 13926251106 240 0 240 13926435656 132 1512 1644 15013685858 3659 3538 7197 15920133257 3156 2936 6092 15989002119 1938 180 2118 18211575961 1527 2106 3633 18320173382 9531 2412 11943 84138413 4116 1432 5548
推荐阅读
-
使用bcompiler对PHP文件进行加密的代码_PHP教程
-
Python中使用pprint函数进行格式化输出的教程
-
使用Python下的XSLT API进行web开发的简单教程
-
php的list()的一步操作给一组变量进行赋值的使用_PHP教程
-
php+mysqli使用预处理技术进行数据库查询的方法，mysqli预处理_PHP教程
-
ThinkPHP整合Smarty模板引擎的使用以及查询方法汇总_PHP教程
-
MySQL的源码安装及使用UDFs进行数据自动更新的教程
-
iOS中使用URL Scheme进行App跳转的教程
-
Python使用email模块对邮件进行编码和解码的实例教程
-
iOS中使用URL Scheme进行App跳转的教程