使用mapreduce进行流量汇总程序开发

现有文件关于流量文件内容如下

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<LongWritable, Text, Text, FlowBean>{
        @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("\\s+");
            //获取手机号
            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<Text, FlowBean, Text, FlowBean>{
        @Override
        protected void reduce(Text key, Iterable<FlowBean> 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