hadoop自定义实现排序流量统计

1.上篇文章讲到mapreduce实现简单的流量统计，但最后的结果是按手机号的字典顺序进行输出的，如果我们需要实现按总流量的大小进行排序输出，怎么办？

2.我们可以用上篇的结果文件作为mapreduce的输入，从新写一个mapreduce程序。

3.map输出为FlowBean作为key，输出为null，而FlowBean我们实现WritableComparable接口，自定义它的排序规则，那么map输出到reduce框架中的key时候，他就会按照我们自定义的规则排序输出。

4.FlowBean具体代码如下：

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;
    }




}
           

5.mapreduce程序代码如下：

package com.zhichao.wan.flowmr.sort;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
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;

import com.zhichao.wan.flowmr.FlowBean;

public class FlowSortMR {


    public static class SortMapper extends Mapper<LongWritable, Text, FlowBean, NullWritable>{

        @Override
        protected void map(LongWritable key, Text value,Context context)
                throws IOException, InterruptedException {
            String line = value.toString();

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

            String phoneNB=fileds[0];
            long u_load=Long.parseLong(fileds[1]);
            long d_load=Long.parseLong(fileds[2]);

            context.write(new FlowBean(phoneNB, u_load, d_load), NullWritable.get());

        }

    }

    public static class sortReducder extends Reducer<FlowBean, NullWritable, Text, FlowBean>{
        @Override
        protected void reduce(FlowBean key, Iterable<NullWritable> values,Context context)
                throws IOException, InterruptedException {

            String phoneNB=key.getPhoneNB();

            context.write(new Text(phoneNB), new FlowBean(phoneNB, key.getU_load(), key.getD_load()));


        }

    }


    public static void main(String[] args) throws Exception {


        Configuration conf=new Configuration();

        Job job=Job.getInstance(conf);


        job.setJarByClass(FlowSortMR.class);

        job.setMapperClass(SortMapper.class);
        job.setReducerClass(sortReducder.class);

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

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


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

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



        job.waitForCompletion(true);



    }


}
           

6.运行结果如下：

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