Hadoop MapReduce（FlowCount） Java编程

7865432

编写PhoneFlow程序,计算手机上行流量、下行流量以及总流量，数据如下：

13685295623 122  201

13985295600 102  11

13885295622 22   101

13785295633 120  20

1、FlowMapper：

package com.hadoop.flow;

import java.io.IOException;

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


public class FlowMapper extends Mapper<LongWritable, Text, Text, FlowBean>{
        /**
         * 数据格式:
         * 13685295623 122  201
         * 13985295600 102  11
         * 13885295622 22   101
         * 13785295633 120  20
         */
        @Override
        protected void map(LongWritable key, Text value,Context context) throws IOException, InterruptedException {
               
                String line=value.toString();
               
                String [] fields=StringUtils.split(line,"\t");
               
                String phoneNB=fields[0];
                long up_flow=Long.valueOf(fields[1]);
                long d_flow=Long.valueOf(fields[2]);
                context.write(new Text(phoneNB), new FlowBean(phoneNB,up_flow,d_flow));
        }

}
2、FlowReducer：
package com.hadoop.flow;

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 upflowC=0;
                long dflowD=0;
               
                for(FlowBean bean:values){
                        upflowC+=bean.getUp_flow();
                        dflowD+=bean.getD_flow();
                }
                context.write(key,new FlowBean(key.toString(),upflowC,dflowD));
               
        }

}

3、FlowRunner
package com.hadoop.flow;

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

public class FlowRunner extends Configured implements Tool{
       
       
        public int run(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.setOutputKeyClass(Text.class);
                job.setOutputValueClass(FlowBean.class);
               
                job.setMapOutputKeyClass(Text.class);
                job.setMapOutputValueClass(FlowBean.class);
               
                FileInputFormat.setInputPaths(job,new Path(args[0]));
                FileOutputFormat.setOutputPath(job,new Path(args[1]));
               
                return job.waitForCompletion(true)?0:1;
        }
       
        public static void main(String[] args) throws Exception {
                ToolRunner.run(new Configuration(), new FlowRunner(), args);
        }

}
4、FlowBean ：
package com.hadoop.flow;

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

import org.apache.hadoop.io.Writable;

public class FlowBean implements Writable{
       
        private String phoneNB;
        private long up_flow;
        private long d_flow;
        private long s_flow;
       
        public FlowBean(){
               
        }
       
        public FlowBean (String phoneNB,long up_flow,long d_flow){
                this.phoneNB=phoneNB;
                this.up_flow=up_flow;
                this.d_flow=d_flow;
                this.s_flow=up_flow+d_flow;
        }
       
        public String getPhoneNB() {
                return phoneNB;
        }

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

        public long getUp_flow() {
                return up_flow;
        }

        public void setUp_flow(long up_flow) {
                this.up_flow = up_flow;
        }

        public long getD_flow() {
                return d_flow;
        }

        public void setD_flow(long d_flow) {
                this.d_flow = d_flow;
        }

        public long getS_flow() {
                return s_flow;
        }

        public void setS_flow(long s_flow) {
                this.s_flow = s_flow;
        }
       
        //
        public void write(DataOutput out) throws IOException {
                out.writeUTF(phoneNB);
                out.writeLong(up_flow);
                out.writeLong(d_flow);
                out.writeLong(s_flow);
        }
               
        public void readFields(DataInput in) throws IOException {
                phoneNB= in.readUTF();
                up_flow=in.readLong();
                d_flow=in.readLong();
                s_flow=in.readLong();
        }
       
        @Override
        public String toString() {
                return up_flow+"   "+d_flow+"   "+"   "+s_flow;
        }
}