/*** @author: PrincessHug* @date: 2019/3/23, 23:38* @Blog: https://www.cnblogs.com/HelloBigTable/*/
public class FlowBean implements Writable {private long upFlow;private long dwFlow;private long flowSum;public long getUpFlow() {return upFlow;}public void setUpFlow(long upFlow) {this.upFlow = upFlow;}public long getDwFlow() {return dwFlow;}public void setDwFlow(long dwFlow) {this.dwFlow = dwFlow;}public long getFlowSum() {return flowSum;}public void setFlowSum(long flowSum) {this.flowSum = flowSum;}public FlowBean() {}public FlowBean(long upFlow, long dwFlow) {this.upFlow = upFlow;this.dwFlow = dwFlow;this.flowSum = upFlow + dwFlow;}/*** 序列化* @param out 输出流* @throws IOException*/@Overridepublic void write(DataOutput out) throws IOException {out.writeLong(upFlow);out.writeLong(dwFlow);out.writeLong(flowSum);}/*** 反序列化* @param in* @throws IOException*/@Overridepublic void readFields(DataInput in) throws IOException {upFlow = in.readLong();dwFlow = in.readLong();flowSum = in.readLong();}@Overridepublic String toString() {return upFlow + "\t" + dwFlow + "\t" + flowSum;}
}public class FlowCountMapper extends Mapper<LongWritable, Text,Text,FlowBean> {@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//获取数据String line = value.toString();//切分数据String[] fields = line.split("\t");//封装数据String phoneNum = fields[1];long upFlow = Long.parseLong(fields[fields.length - 3]);long dwFlow = Long.parseLong(fields[fields.length - 2]);//发送数据context.write(new Text(phoneNum),new FlowBean(upFlow,dwFlow));}
}public class FlowCountReducer extends Reducer<Text,FlowBean,Text,FlowBean> {@Overrideprotected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {//聚合数据long upFlow_sum = 0;long dwFlow_sum = 0;for (FlowBean f:values){upFlow_sum += f.getUpFlow();dwFlow_sum += f.getDwFlow();}//发送数据context.write(key,new FlowBean(upFlow_sum,dwFlow_sum));}
}public class FlowPartitioner extends Partitioner<Text,FlowBean> {@Overridepublic int getPartition(Text key, FlowBean value, int i) {//获取用来分区的电话号码前三位String phoneNum = key.toString().substring(0, 3);//设置分区逻辑int partitionNum = 4;if ("135".equals(phoneNum)){return 0;}else if ("137".equals(phoneNum)){return 1;}else if ("138".equals(phoneNum)){return 2;}else if ("139".equals(phoneNum)){return 3;}return partitionNum;}
}
public class FlowCountDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {//获取配置，定义工具Configuration conf = new Configuration();Job job = Job.getInstance();//设置运行类job.setJarByClass(FlowCountDriver.class);//设置Mapper类及Mapper输出数据类型job.setMapperClass(FlowCountMapper.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(FlowBean.class);//设置Reducer类及其输出数据类型job.setReducerClass(FlowCountReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(FlowBean.class);//设置自定义分区job.setPartitionerClass(FlowPartitioner.class);job.setNumReduceTasks(5);//设置文件输入输出流FileInputFormat.setInputPaths(job,new Path("G:\\mapreduce\\flow\\in"));FileOutputFormat.setOutputPath(job,new Path("G:\\mapreduce\\flow\\inpartitionout"));//返回运行完成if (job.waitForCompletion(true)){System.out.println("运行完毕！");}else {System.out.println("运行出错!");}}
}