文档章节

用mapreduce处理用户使用流量统计

键盘上跳舞
 键盘上跳舞
发布于 2017/03/11 14:16
字数 1221
阅读 70
收藏 0

需求:通过日志分析将用户每月的使用流量情况汇总,为推荐系统系统数据服务

 设计:将日志数据flow.log 通过map task读取文本数据,以电话号码为key,把用户的详细数据构造bean对象,交由reduce task 将数据汇总。

    为了测试,先模拟实现,代码如下:

    FlowBean.java

    

/**
 * 
 * @ClassName FlowBean
 * @author liukang
 * @Description TODO 流量bean
 * @Date 2017年3月10日 下午10:26:48
 * @version 1.0.0
 */
public class FlowBean implements Writable
{ // 上行流量
    private long upFlow;
    
    // 下行流量
    private long downFlow;
    
    // 总流量
    private long sumFlow;
    
    // 序列化时,需要反射调用空的无参数构造
    public FlowBean()
    {
        super();
        // TODO Auto-generated constructor stub
    }
    
    public FlowBean(long upFlow, long downFlow)
    {
        super();
        this.upFlow = upFlow;
        this.downFlow = downFlow;
        this.sumFlow = upFlow + downFlow;
    }
    
    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;
    }
    
    /*
     * 序列化
     * 
     * @see org.apache.hadoop.io.Writable#write(java.io.DataOutput)
     */
    @Override
    public void write(DataOutput out)
        throws IOException
    {
        out.writeLong(upFlow);
        out.writeLong(downFlow);
        out.writeLong(sumFlow);
    }
    
    /*
     * 反序列化时
     * 
     * 顺序:和序列化的顺序一致
     * 
     * @see org.apache.hadoop.io.Writable#readFields(java.io.DataInput)
     */
    @Override
    public void readFields(DataInput in)
        throws IOException
    {
        upFlow = in.readLong();
        downFlow = in.readLong();
        sumFlow = in.readLong();
    }
    
    /*
     * 重写toString方法 Description:
     * 
     * @see java.lang.Object#toString()
     */
    @Override
    public String toString()
    {
        return upFlow + " \t" + downFlow + "\t" + sumFlow;
    }
    
}

        FlowCount.java

            

/**
 * 统计用户上行流量、下行流量、总流量
 * 
 * @ClassName FlowCount
 * @author liukang
 * @Description TODO
 * @Date 2017年3月10日 下午10:41:38
 * @version 1.0.0
 */
public class FlowCount
{
    /**
     * map模块
     * 
     * @ClassName FlowMapper
     * @author liukang
     * @Description KEYIN, VALUEIN KEYIN 为mapreduce读取一行文本的其实偏移量, VALUEIN为读取一行文本的内容
     * @Date 2017年3月10日 下午10:40:18
     * @version 1.0.0
     */
    static class FlowCountMapper extends Mapper<LongWritable, Text, Text, FlowBean>
    {
        @Override
        protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException
        {
            // 读取一行文本
            String line = value.toString();
            // 切分读取的文本内容
            String[] fields = line.split("\t");
            // 获取手机号
            String phoneNum = fields[1];
            // 获取上行流量和下行流量
            long upFlow = Long.parseLong(fields[fields.length - 3]);
            long downFlow = Long.parseLong(fields[fields.length - 2]);
            // 将统计的结果输出,交给reduce
            context.write(new Text(phoneNum), new FlowBean(upFlow, downFlow));
        }
        
    }
    
    /**
     * reduce 模块
     * 
     * @ClassName FlowReducer
     * @author liukang
     * @Description TODO reduce 输出结果经过序列化和反序列化为FlowBean 对象输出到文本内容中
     * @Date 2017年3月10日 下午10:58:57
     * @version 1.0.0
     */
    static class FlowCountReducer extends Reducer<Text, FlowBean, Text, FlowBean>
    {
        /*
         * 经过map阶段reduce得到的入参结构:<183323,bean1><183323,bean2><183323,bean3><183323,bean4>....... Description:
         * 对于reduce来说(对于单独的一个reduce task),得到的是一组数据结构相同的数据,所在在其操作阶段就是对拿到的集合操作
         * 
         * @see org.apache.hadoop.mapreduce.Reducer#reduce(KEYIN, java.lang.Iterable,
         * org.apache.hadoop.mapreduce.Reducer.Context)
         */
        @Override
        protected void reduce(Text phoneNum, Iterable<FlowBean> iterable, Context context)
            throws IOException, InterruptedException
        {
            long sum_upFlow = 0;
            long sum_downFlow = 0;
            // 循环遍历入参,将流量统计
            for (FlowBean flowBean : iterable)
            {
                sum_upFlow += flowBean.getUpFlow();
                sum_downFlow += flowBean.getDownFlow();
            }
            
            context.write(phoneNum, new FlowBean(sum_upFlow, sum_downFlow));
        }
    }
    
    // 总任务调度
    public static void main(String[] args)
        throws IOException, ClassNotFoundException, InterruptedException
    {
        
        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);
    }
}

        flow.log模拟数据

        

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

执行结果:

    

 

 

 

 

 

 

 

 

 

 

 

© 著作权归作者所有

键盘上跳舞
粉丝 19
博文 68
码字总数 66382
作品 0
海淀
程序员
私信 提问
大数据教程(9.1)流量汇总排序的mr实现

上一章我们有讲到一个mapreduce案例——移动流量排序,如果我们要将最后的输出结果按总流量大小逆序输出,该怎么实现呢?本节博主将分享这个实现的过程。 一、分析 首先,要实现这个功能,我...

em_aaron
2018/12/01
59
0
使用Spark Streaming SQL进行PV/UV统计

作者:关文选,花名云魄,阿里云E-MapReduce 高级开发工程师,专注于流式计算,Spark Contributor 1.背景介绍 PV/UV统计是流式分析一个常见的场景。通过PV可以对访问的网站做流量或热点分析,...

开源大数据EMR
前天
0
0
7个实例全面掌握Hadoop MapReduce

作者介绍 杜亦舒,创业中,技术合伙人,喜欢研究分享技术。个人订阅号:性能与架构。 本文旨在帮您快速了解 MapReduce 的工作机制和开发方法,解决以下几个问题: 文章中提供了程序实例中涉及...

杜亦舒
2017/06/08
0
0
大数据(hadoop-Mapreduce原理架构)

课程目标: 1:MapReduce的应用场景 2:MapReduce编程模型 3:MapReduce的架构 4:常见MapReduce应用场景 5:总结 MapReduce的定义 源自于Google的MapReduce论文 发表于2004年12月 Hadoop M...

这很耳东先生
04/30
40
0
mongodb的多表关联哪种处理方式最优

@huzorro 你好,想跟你请教个问题:这段时间一直在看您发表的mongodb方面的文章,对我的帮助的很大,非常感谢!今天遇到了一个问题,想了一下午,查看了很多资料也没能找到合理的解决方式,不...

重阳真人
2013/03/23
9.9K
5

没有更多内容

加载失败,请刷新页面

加载更多

nginx学习笔记

中间件位于客户机/ 服务器的操作系统之上,管理计算机资源和网络通讯。 是连接两个独立应用程序或独立系统的软件。 web请求通过中间件可以直接调用操作系统,也可以经过中间件把请求分发到多...

码农实战
今天
5
0
Spring Security 实战干货:玩转自定义登录

1. 前言 前面的关于 Spring Security 相关的文章只是一个预热。为了接下来更好的实战,如果你错过了请从 Spring Security 实战系列 开始。安全访问的第一步就是认证(Authentication),认证...

码农小胖哥
今天
12
0
JAVA 实现雪花算法生成唯一订单号工具类

import lombok.SneakyThrows;import lombok.extern.slf4j.Slf4j;import java.util.Calendar;/** * Default distributed primary key generator. * * <p> * Use snowflake......

huangkejie
昨天
12
0
PhotoShop 色调:RGB/CMYK 颜色模式

一·、 RGB : 三原色:红绿蓝 1.通道:通道中的红绿蓝通道分别对应的是红绿蓝三种原色(RGB)的显示范围 1.差值模式能模拟三种原色叠加之后的效果 2.添加-颜色曲线:调整图像RGB颜色----R色增强...

东方墨天
昨天
11
1
将博客搬至CSDN

将博客搬至CSDN

算法与编程之美
昨天
13
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部