package com.ares.hadoop.mr.flowgroup;import java.io.DataInput;import java.io.DataOutput;import java.io.IOException;import org.apache.hadoop.io.WritableComparable;public class FlowBean implements WritableComparable<FlowBean>{    private String phoneNum;    private long upFlow;    private long downFlow;    private long sumFlow;        public FlowBean() {        // TODO Auto-generated constructor stub    }    //    public FlowBean(String phoneNum, long upFlow, long downFlow, long sumFlow) {//        super();//        this.phoneNum = phoneNum;//        this.upFlow = upFlow;//        this.downFlow = downFlow;//        this.sumFlow = sumFlow;//    }    public String getPhoneNum() {        return phoneNum;    }    public void setPhoneNum(String phoneNum) {        this.phoneNum = phoneNum;    }    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;    }    @Override    public void readFields(DataInput in) throws IOException {        // TODO Auto-generated method stub        phoneNum = in.readUTF();        upFlow = in.readLong();        downFlow = in.readLong();        sumFlow = in.readLong();    }    @Override    public void write(DataOutput out) throws IOException {        // TODO Auto-generated method stub        out.writeUTF(phoneNum);        out.writeLong(upFlow);        out.writeLong(downFlow);        out.writeLong(sumFlow);    }    @Override    public String toString() {        return "" + phoneNum + "\t" + upFlow + "\t" + downFlow + "\t" + sumFlow;    }    @Override    public int compareTo(FlowBean o) {        // TODO Auto-generated method stub        return sumFlow>o.getSumFlow()?-1:1;    }    }

package com.ares.hadoop.mr.flowgroup;import java.io.IOException;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.conf.Configured;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 org.apache.hadoop.util.StringUtils;import org.apache.hadoop.util.Tool;import org.apache.hadoop.util.ToolRunner;import org.apache.log4j.Logger;import com.ares.hadoop.mr.exception.LineException;import com.ares.hadoop.mr.flowgroup.FlowBean;;public class FlowGroup extends Configured implements Tool {    private static final Logger LOGGER = Logger.getLogger(FlowGroup.class);    enum Counter {        LINESKIP    }        public static class FlowGroupMapper extends Mapper<LongWritable, Text,         Text, FlowBean> {        private String line;        private int length;        private final static char separator = ‘\t‘;                private String phoneNum;        private long upFlow;        private long downFlow;        //private long sumFlow;                private Text text = new Text();        private FlowBean flowBean = new FlowBean();                @Override        protected void map(                LongWritable key,                Text value,                Mapper<LongWritable, Text, Text, FlowBean>.Context context)                throws IOException, InterruptedException {            // TODO Auto-generated method stub            //super.map(key, value, context);            String errMsg;            try {                line = value.toString();                String[] fields = StringUtils.split(line, separator);                length = fields.length;                if (length != 11) {                    throw new LineException(key.get() + ", " + line + " LENGTH INVALID, IGNORE...");                }                                phoneNum = fields[1];                upFlow = Long.parseLong(fields[length-3]);                downFlow = Long.parseLong(fields[length-2]);                //sumFlow = upFlow + downFlow;                                text.set(phoneNum);                flowBean.setPhoneNum(phoneNum);                flowBean.setUpFlow(upFlow);                flowBean.setDownFlow(downFlow);                //flowBean.setSumFlow(sumFlow);                                context.write(text, flowBean);            } catch (LineException e) {                // TODO: handle exception                LOGGER.error(e);                System.out.println(e);                context.getCounter(Counter.LINESKIP).increment(1);                return;            } catch (NumberFormatException e) {                // TODO: handle exception                errMsg = key.get() + ", " + line + " FLOW DATA INVALID, IGNORE...";                LOGGER.error(errMsg);                System.out.println(errMsg);                context.getCounter(Counter.LINESKIP).increment(1);                return;            } catch (Exception e) {                // TODO: handle exception                LOGGER.error(e);                System.out.println(e);                context.getCounter(Counter.LINESKIP).increment(1);                return;            }                    }    }        public static class FlowGroupReducer extends Reducer<Text, FlowBean,         FlowBean, NullWritable> {                private FlowBean flowBean = new FlowBean();                @Override        protected void reduce(                Text key,                Iterable<FlowBean> values,                Reducer<Text, FlowBean, FlowBean, NullWritable>.Context context)                throws IOException, InterruptedException {            // TODO Auto-generated method stub            //super.reduce(arg0, arg1, arg2);            long upFlowCounter = 0;            long downFlowCounter = 0;                        for (FlowBean flowBean : values) {                upFlowCounter += flowBean.getUpFlow();                downFlowCounter += flowBean.getDownFlow();            }            flowBean.setPhoneNum(key.toString());            flowBean.setUpFlow(upFlowCounter);            flowBean.setDownFlow(downFlowCounter);            flowBean.setSumFlow(upFlowCounter + downFlowCounter);                        context.write(flowBean, NullWritable.get());        }    }    @Override    public int run(String[] args) throws Exception {        // TODO Auto-generated method stub        String errMsg = "FlowGroup: TEST STARTED...";        LOGGER.debug(errMsg);        System.out.println(errMsg);                Configuration conf = new Configuration();        //FOR Eclipse JVM Debug          //conf.set("mapreduce.job.jar", "flowsum.jar");        Job job = Job.getInstance(conf);                // JOB NAME        job.setJobName("FlowGroup");                // JOB MAPPER & REDUCER        job.setJarByClass(FlowGroup.class);        job.setMapperClass(FlowGroupMapper.class);        job.setReducerClass(FlowGroupReducer.class);                // JOB PARTITION        job.setPartitionerClass(FlowGroupPartition.class);                // JOB REDUCE TASK NUMBER        job.setNumReduceTasks(5);                // MAP & REDUCE        job.setOutputKeyClass(FlowBean.class);        job.setOutputValueClass(NullWritable.class);        // MAP        job.setMapOutputKeyClass(Text.class);        job.setMapOutputValueClass(FlowBean.class);                // JOB INPUT & OUTPUT PATH        //FileInputFormat.addInputPath(job, new Path(args[0]));        FileInputFormat.setInputPaths(job, args[1]);        FileOutputFormat.setOutputPath(job, new Path(args[2]));                // VERBOSE OUTPUT        if (job.waitForCompletion(true)) {            errMsg = "FlowGroup: TEST SUCCESSFULLY...";            LOGGER.debug(errMsg);            System.out.println(errMsg);            return 0;        } else {            errMsg = "FlowGroup: TEST FAILED...";            LOGGER.debug(errMsg);            System.out.println(errMsg);            return 1;        }                        }        public static void main(String[] args) throws Exception {        if (args.length != 3) {            String errMsg = "FlowGroup: ARGUMENTS ERROR";            LOGGER.error(errMsg);            System.out.println(errMsg);            System.exit(-1);        }                int result = ToolRunner.run(new Configuration(), new FlowGroup(), args);        System.exit(result);    }}

package com.ares.hadoop.mr.flowgroup;import java.util.HashMap;import org.apache.hadoop.mapreduce.Partitioner;public class FlowGroupPartition<KEY, VALUE> extends Partitioner<KEY, VALUE>{    private static HashMap<String, Integer> groupMap = new HashMap<String, Integer>();    static {        groupMap.put("135", 0);        groupMap.put("136", 1);        groupMap.put("137", 2);        groupMap.put("138", 3);    }        @Override    public int getPartition(KEY key, VALUE value, int numPartitions) {        // TODO Auto-generated method stub        return (groupMap.get(key.toString().substring(0, 3)) == null)?4:            groupMap.get(key.toString().substring(0, 3));    }}