MapReduce统计每个用户的使用总流量

package com.zy.flow;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;public class Flow implements Writable{private Text phone;private LongWritable upflow;private LongWritable downflow;private LongWritable sumflow;//这个对象以后要在集群中传输&＃xff0c;所以要可序列化//序列化反序列化顺序要一致&＃64;Override//反序列化时会调用该方法public void readFields(DataInput in) throws IOException {phone&＃61;new Text(in.readUTF());upflow&＃61;new LongWritable(in.readLong());downflow&＃61;new LongWritable(in.readLong());sumflow&＃61;new LongWritable(in.readLong());}&＃64;Override//序列化时会调用该方法public void write(DataOutput out) throws IOException {out.writeUTF(phone.toString());out.writeLong(upflow.get());out.writeLong(downflow.get());out.writeLong(sumflow.get());}public Text getPhone() {return phone;}public void setPhone(Text phone) {this.phone &＃61; phone;}public LongWritable getUpflow() {return upflow;}public void setUpflow(LongWritable upflow) {this.upflow &＃61; upflow;}public LongWritable getDownflow() {return downflow;}public void setDownflow(LongWritable downflow) {this.downflow &＃61; downflow;}public LongWritable getSumflow() {return sumflow;}public void setSumflow(LongWritable sumflow) {this.sumflow &＃61; sumflow;}public Flow() {}public Flow(Text phone, LongWritable upflow, LongWritable downflow, LongWritable sumflow) {super();this.phone &＃61; phone;this.upflow &＃61; upflow;this.downflow &＃61; downflow;this.sumflow &＃61; sumflow;}public Flow(LongWritable upflow, LongWritable downflow, LongWritable sumflow) {super();this.upflow &＃61; upflow;this.downflow &＃61; downflow;this.sumflow &＃61; sumflow;}&＃64;Overridepublic String toString() {return upflow&＃43;"\t"&＃43;downflow&＃43;"\t"&＃43;sumflow;}}

package com.zy.flow;import java.io.IOException;import javax.security.auth.callback.LanguageCallback;import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;public class FlowMap extends Mapper{&＃64;Overrideprotected void map(LongWritable key, Text value, Mapper.Context context)throws IOException, InterruptedException {//输入的值 value//切分value 寻找有价值的列String[] split &＃61; value.toString().split("\t");int length&＃61;split.length;//取哪几列split[1] split[length-3] split[length-2]String phone&＃61;split[1];Long upflow&＃61;Long.parseLong(split[length-3]);Long downflow&＃61;Long.parseLong(split[length-2]);Long sumflow&＃61;upflow&＃43;downflow;//输出context.write(new Text(phone), new Flow(new Text(phone), new LongWritable(upflow), new LongWritable(downflow),new LongWritable(sumflow)));//对象里虽然用不到phone但是要给它赋值&＃xff0c;不然序列化时会报空指针异常
}
}

package com.zy.flow;
import java.util.HashMap;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;public class Part extends Partitioner {//分区//逻辑自己写
HashMap map &＃61; new HashMap();public void setMap(){map.put("135",0);map.put("136", 1);map.put("137",2);map.put("138", 3);map.put("139",4);} &＃64;Overridepublic int getPartition(Text key, Flow value, int arg2) {setMap();//从输入的数据中获得电话的前三位跟map对比。决定分到哪个区中String substring &＃61; key.toString().substring(0, 3);//例如截取135return map.get(substring)&＃61;&＃61;null?5:map.get(substring);//根据键取值 键135 取出0 //其他号码分到&＃xff08;编号为5&＃xff09;第6个区中
}//在这个逻辑下partition分了6个区&＃xff0c;所以以后要指定6个reducetask

}

package com.zy.flow;import java.io.IOException;import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;public class FlowReduce extends Reducer{&＃64;Overrideprotected void reduce(Text key, Iterable value, Reducer.Context context)throws IOException, InterruptedException {//累加long allup&＃61;0;long alldown&＃61;0;for (Flow flow : value) {allup&＃43;&＃61;Long.parseLong(flow.getUpflow().toString());alldown&＃43;&＃61;Long.parseLong(flow.getDownflow().toString());}long allsum&＃61;allup&＃43;alldown;context.write(key, new Flow(new Text(key), new LongWritable(allup), new LongWritable(alldown), new LongWritable(allsum)));}}

package com.zy.flow;import java.io.IOException;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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;public class FlowApp {public static void main(String[] args) throws Exception {//创建配置对象Configuration configuration &＃61; new Configuration();//得到job实例Job job &＃61; Job.getInstance(configuration);//指定job运行类job.setJarByClass(FlowApp.class);//指定job中的mapperjob.setMapperClass(FlowMap.class);//指定mapper中的输出键和值类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(Flow.class);//指定job中的reducerjob.setReducerClass(FlowReduce.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(Flow.class);//-----//指定Partitioner使用的类job.setPartitionerClass(Part.class);//指定ReduceTask数量job.setNumReduceTasks(6);//-----//指定输入文件FileInputFormat.setInputPaths(job, new Path(args[0]));//运行时填入参数//指定输出文件FileOutputFormat.setOutputPath(job, new Path(args[1]));//提交作业job.waitForCompletion(true);}}