public FlowBean() {
	super();
}

WritableComparable<BinaryComparable>

package com.tangxiaocong.mapreduce.writable;
 
import org.apache.hadoop.io.Writable;
 
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
 
/*
*
* 定義bean類
* 需要實現(xiàn)writable
* 重寫序列化和反序列化方法
* 重寫空參構(gòu)造
* 重寫tostring方法
*
* */
public class FlowBean  implements Writable {
   private  long upFlow;
    private  long downFlow;
    private  long sumFlow;
 
    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;
    }
    public void setSumFlow() {
        this.sumFlow = this.downFlow+this.upFlow;
    }
    //生成空參構(gòu)造函數(shù)由于反射  快捷鍵alt   + insert
 
    public FlowBean() {
    }
 
    @Override
    public void write(DataOutput out) throws IOException {
        //序列化方法
        //  向緩沖流中寫入Long類型的數(shù)據(jù)
        out.writeLong(upFlow);
        out.writeLong(downFlow);
        out.writeLong(sumFlow);
    }
 
    @Override
    public void readFields(DataInput in) throws IOException {
//反序列化方法
        //讀取緩沖區(qū)中數(shù)據(jù)
        this.upFlow= in.readLong();
        this.downFlow= in.readLong();
        this.sumFlow= in.readLong();
    }
 
    @Override
    public String toString() {
        return upFlow + "\t"+downFlow +"\t"+ sumFlow ;
    }
}

package com.tangxiaocong.mapreduce.writable;
 
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
 
import java.io.IOException;
 
public class FlowMapper extends Mapper<LongWritable, Text,Text,FlowBean> {
 
    private Text outK=new Text();
    private  FlowBean outV=new FlowBean();  //調(diào)用的無參構(gòu)造函數(shù)
 
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, FlowBean>.Context context) throws IOException, InterruptedException {
        //1 獲取一行
        //1	13736230513	192.196.100.1	www.atguigu.com	2481	24681	200
 
        String s = value.toString();// 將數(shù)據(jù)轉(zhuǎn)換成string
        //2 進行切割
 
        String[] split = s.split("\t"); //將數(shù)據(jù)按寫入形式進行切割
        //3 抓取想要的數(shù)據(jù)
        //根據(jù)角標獲取  手機號  上行流量  下行流量
 
 
        String phone = split[1];
        String up = split[split.length - 3];//  不能正序 因為有的屬性是沒有字段的
        String down = split[split.length - 2];
//     封裝輸出的kv
 
        outK.set(phone);
        outV.setUpFlow(Long.parseLong(up));//  up為string類型
        outV.setDownFlow(Long.parseLong(down));
        outV.setSumFlow();          //
 
        //寫出
        context.write(outK,outV);
    }
}

package com.tangxiaocong.mapreduce.writable;
 
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
 
import java.io.IOException;
 
public class FlowReducer extends Reducer  <Text,FlowBean,Text,FlowBean>{
   private FlowBean outv=new FlowBean();
 
 
    @Override
    protected void reduce(Text key, Iterable<FlowBean> values, Reducer<Text, FlowBean, Text, FlowBean>.Context context) throws IOException, InterruptedException {
 
         long totalUp=0;
         long totaldown=0;
         
        //分析   傳入TEXT  為手機號  后邊為集合（Bean類的對象的集合）輸出還是一個一個bean類  （每個手機號的總和）
        for (FlowBean value : values) {  //傳入的參數(shù)是同一個key的
            totalUp+=value.getUpFlow();
            totaldown+=value.getDownFlow();
        }
        //  現(xiàn)在求出的是每個手機號的總的上行流量  下行流量
            //封裝  key不需要
        //outv
    outv.setUpFlow(totalUp);
    outv.setDownFlow(totaldown);
    outv.setSumFlow();
    //寫出
        context.write(key,outv);
    }
}

package com.tangxiaocong.mapreduce.writable;
 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 
import java.io.IOException;
 
public class FlowDriver {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        //獲取JOB
        Configuration entries = new Configuration();
        Job job = Job.getInstance(entries);
 
        job.setJarByClass(FlowDriver.class);
        //關(guān)聯(lián)mapper  和reduce
        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReducer.class);
 
        //設(shè)置mapper  輸出的key 和value
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);
 
        // 設(shè)置最終的數(shù)據(jù)輸出的key和value 類型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);
 
        //設(shè)置數(shù)據(jù)的輸入路徑和輸出路徑
        FileInputFormat.setInputPaths(job, new Path("D:\\hadoop\\phone_data.txt"));
        FileOutputFormat.setOutputPath(job, new Path("D:\\hadoop\\output3"));
        //提交job
        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);
 
    }
 
}