Hadoop MapReduce（FlowCount） Java编程

雷锋

　　编写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>　　/**
　　* 数据格式:
　　* 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>　　@Override
　　protected void reduce(Text key, Iterable 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>　　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>　　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;
　　}
　　}