1 package mapreduce;
  2
  3 import java.io.DataInput;
  4 import java.io.DataOutput;
  5 import java.io.IOException;
  6
  7 import org.apache.hadoop.conf.Configuration;
  8 import org.apache.hadoop.fs.Path;
  9 import org.apache.hadoop.io.LongWritable;
 10 import org.apache.hadoop.io.Text;
 11 import org.apache.hadoop.io.Writable;
 12 import org.apache.hadoop.mapreduce.Job;
 13 import org.apache.hadoop.mapreduce.Mapper;
 14 import org.apache.hadoop.mapreduce.Reducer;
 15 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
 16 import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
 17 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 18 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
 19 import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
 20
 21 public class KpiApp {
 22     static final String INPUT_PATH = "hdfs://hadoop:9000/wlan";
 23     static final String OUT_PATH = "hdfs://hadoop:9000/out";
 24     public static void main(String[] args) throws Exception{
 25         final Job job = new Job(new Configuration(), KpiApp.class.getSimpleName());
 26
 27         FileInputFormat.setInputPaths(job, INPUT_PATH);//1.1 指定输入文件路径
 28
 29         job.setInputFormatClass(TextInputFormat.class);//指定哪个类用来格式化输入文件
 30
 31         job.setMapperClass(MyMapper.class);//1.2指定自定义的Mapper类
 32
 33         job.setMapOutputKeyClass(Text.class);//指定输出<k2,v2>的类型
 34         job.setMapOutputValueClass(KpiWritable.class);
 35
 36         job.setPartitionerClass(HashPartitioner.class);//1.3 指定分区类
 37         job.setNumReduceTasks(1);
 38
 39         //1.4 TODO 排序、分区
 40
 41         //1.5  TODO （可选）合并
 42
 43         job.setReducerClass(MyReducer.class);//2.2 指定自定义的reduce类
 44
 45         job.setOutputKeyClass(Text.class);//指定输出<k3,v3>的类型
 46         job.setOutputValueClass(KpiWritable.class);
 47
 48         FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));//2.3 指定输出到哪里
 49
 50         job.setOutputFormatClass(TextOutputFormat.class);//设定输出文件的格式化类
 51
 52         job.waitForCompletion(true);//把代码提交给JobTracker执行
 53     }
 54
 55     static class MyMapper extends Mapper<LongWritable, Text, Text, KpiWritable>{
 56         protected void map(LongWritable key, Text value, org.apache.hadoop.mapreduce.Mapper<LongWritable,Text,Text,KpiWritable>.Context context) throws IOException ,InterruptedException {
 57             final String[] splited = value.toString().split("\t");
 58             final String msisdn = splited[1];
 59             final Text k2 = new Text(msisdn);
 60             final KpiWritable v2 = new KpiWritable(splited[6],splited[7],splited[8],splited[9]);
 61             context.write(k2, v2);
 62         };
 63     }
 64
 65     static class MyReducer extends Reducer<Text, KpiWritable, Text, KpiWritable>{
 66         /**
 67          * @param    k2    表示整个文件中不同的手机号码
 68          * @param    v2s    表示该手机号在不同时段的流量的集合
 69          */
 70         protected void reduce(Text k2, java.lang.Iterable<KpiWritable> v2s, org.apache.hadoop.mapreduce.Reducer<Text,KpiWritable,Text,KpiWritable>.Context context) throws IOException ,InterruptedException {
 71             long upPackNum = 0L;
 72             long downPackNum = 0L;
 73             long upPayLoad = 0L;
 74             long downPayLoad = 0L;
 75
 76             for (KpiWritable kpiWritable : v2s) {
 77                 upPackNum += kpiWritable.upPackNum;
 78                 downPackNum += kpiWritable.downPackNum;
 79                 upPayLoad += kpiWritable.upPayLoad;
 80                 downPayLoad += kpiWritable.downPayLoad;
 81             }
 82
 83             final KpiWritable v3 = new KpiWritable(upPackNum+"", downPackNum+"", upPayLoad+"", downPayLoad+"");
 84             context.write(k2, v3);
 85         };
 86     }
 87 }
 88
 89 class KpiWritable implements Writable{
 90     long upPackNum;
 91     long downPackNum;
 92     long upPayLoad;
 93     long downPayLoad;
 94
 95     public KpiWritable(){}
 96
 97     public KpiWritable(String upPackNum, String downPackNum, String upPayLoad, String downPayLoad){
 98         this.upPackNum = Long.parseLong(upPackNum);
 99         this.downPackNum = Long.parseLong(downPackNum);
100         this.upPayLoad = Long.parseLong(upPayLoad);
101         this.downPayLoad = Long.parseLong(downPayLoad);
102     }
103
104
105     @Override
106     public void readFields(DataInput in) throws IOException {
107         this.upPackNum = in.readLong();
108         this.downPackNum = in.readLong();
109         this.upPayLoad = in.readLong();
110         this.downPayLoad = in.readLong();
111     }
112
113     @Override
114     public void write(DataOutput out) throws IOException {
115         out.writeLong(upPackNum);
116         out.writeLong(downPackNum);
117         out.writeLong(upPayLoad);
118         out.writeLong(downPayLoad);
119     }
120
121     @Override
122     public String toString() {
123         return upPackNum + "\t" + downPackNum + "\t" + upPayLoad + "\t" + downPayLoad;
124     }
125 }