mapreduce实现简单的流量统计功能

1.数据格式如下：

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

数据的第二列表示手机号，数据的第7列表示上行流量，第8列表示下行流量，mapreduce需要实现，按手机号统计总的上行流量和下行流量，总的流量。

2.新建一个javaBean:

package com.zhichao.wan.flowmr;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;

public class FlowBean implements WritableComparable<FlowBean>{


    private String phoneNB;
    private long u_load;
    private long d_load;
    private long s_load;


    public FlowBean(){

    }

    public FlowBean(String phoneNB, long u_load, long d_load) {
        super();
        this.phoneNB = phoneNB;
        this.u_load = u_load;
        this.d_load = d_load;
        this.s_load=u_load+d_load;
    }

    public String getPhoneNB() {
        return phoneNB;
    }

    public void setPhoneNB(String phoneNB) {
        this.phoneNB = phoneNB;
    }

    public long getU_load() {
        return u_load;
    }

    public void setU_load(long u_load) {
        this.u_load = u_load;
    }

    public long getD_load() {
        return d_load;
    }

    public void setD_load(long d_load) {
        this.d_load = d_load;
    }

    public long getS_load() {
        return s_load;
    }

    public void setS_load(long s_load) {
        this.s_load = s_load;
    }

    @Override
    public void write(DataOutput out) throws IOException {

        out.writeUTF(phoneNB);
        out.writeLong(d_load);
        out.writeLong(u_load);
        out.writeLong(s_load);

    }

    @Override
    public void readFields(DataInput in) throws IOException {


        phoneNB=in.readUTF();
        d_load=in.readLong();
        u_load=in.readLong();
        s_load=in.readLong();

    }

    @Override
    public String toString() {
        return ""+d_load+"\t"+u_load+"\t"+s_load;
    }

    @Override
    public int compareTo(FlowBean arg0) {
        // TODO Auto-generated method stub
        return s_load>arg0.getS_load()?-1:1;
    }




}

3.Map:

package com.zhichao.wan.flowmr;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class FlowMapper extends Mapper<LongWritable, Text, Text, FlowBean>{

    @Override
    protected void map(LongWritable key, Text value,Context context)
            throws IOException, InterruptedException {

        String line = value.toString();

        String[] strings = StringUtils.split(line,"\t");

        String phoneNB=strings[1];      
        String u_load=strings[7];
        String d_load=strings[8];

        context.write(new Text(phoneNB), new FlowBean(phoneNB, Long.parseLong(u_load), Long.parseLong(d_load)));





    }

}

4.reduce:

package com.zhichao.wan.flowmr;

import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class FlowReducer extends Reducer<Text, FlowBean, Text, FlowBean>{

    @Override
    protected void reduce(Text key, Iterable<FlowBean> values,Context context)
            throws IOException, InterruptedException {

        long u_load = 0;
        long d_load = 0;

        for (FlowBean flowBean : values) {

            u_load+=flowBean.getU_load();
            d_load+=flowBean.getD_load();   

        }

        context.write(key, new FlowBean(key.toString(), u_load, d_load));


    }


}

5.Runner:

package com.zhichao.wan.flowmr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class FlowRunner {

    public static void main(String[] args) throws Exception {
        Configuration conf=new Configuration();
        Job job=Job.getInstance(conf);

        job.setJarByClass(FlowRunner.class);

        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);

        FileInputFormat.setInputPaths(job, new Path("F://hadoop/flow/input"));

        FileOutputFormat.setOutputPath(job, new Path("F://hadoop/flow/output"));

        job.waitForCompletion(true);



    }

}

6.运行结果：

13480253104 200 180 380
13502468823 7335    102 7437
13560436666 200 954 1154
13560439658 400 5892    6292
13602846565 1938    12  1950
13660577991 6960    9   6969
13719199419 200 0   200
13726230503 24681   2481    27162
13726238888 24681   2481    27162
13760778710 200 120 320
13826544101 200 0   200
13922314466 3720    3008    6728
13925057413 11058   63  11121
13926251106 200 0   200
13926435656 200 1512    1712
15013685858 3659    27  3686
15920133257 3156    20  3176
15989002119 1938    3   1941
18211575961 1527    12  1539
18320173382 9531    18  9549
84138413    1432    4116    5548

mapreduce实现简单的流量统计功能

继续阅读

【51CTO学院三周年】自学路上的伴侣

在线教育巨头多邻国Duolingo入华一周年，中国市场马力全开

【分类算法】什么是分类算法定义分类与聚类分类过程方法

申请评分模型拒绝推断（RI）方法申请评分模型拒绝推断（RI）方法

专家访谈：搜索开源力量：Lucene技术前景

Sql优化一：sql语句优化

Nacos 2.0 升级前后性能对比压测

尚硅谷—韩顺平—图解 Java设计模式（结构型）（55～）

Storm编译打包过程中遇到的一些问题及解决方法

MapReduce的几个企业级经典面试案例MapReduce的几个企业级经典面试案例

9.spark Core 进阶2--Cashe

浅谈企业活动中进行数据分析的重要性

Ambari介绍和架构原理

详解STM32单片机的堆栈

NOSQL安全攻击

win10本地scala和spark安装安装scala安装spark