一次性完成,统计和排序

我们看下继承的reduce

image.png

reduce需要调用run方法,run方法中不仅执行了reduce最后还执行了cleanup
因为map不断的提交给reduce,reduce排序好了就要写,但是这时候一旦写到文件中,后面再来任务,再写的话,就不能和前面一起排序了

所以我们写到一个treeMap中,然后在cleanup中做treeMap做排序

代码主要把继承reduce中的那个类改了下

  1. package com.folwsum;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.fs.Path;
  5. import org.apache.hadoop.io.LongWritable;
  6. import org.apache.hadoop.io.Text;
  7. import org.apache.hadoop.mapreduce.Job;
  8. import org.apache.hadoop.mapreduce.Mapper;
  9. import org.apache.hadoop.mapreduce.Reducer;
  10. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  11. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  12. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  13. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  14. import org.apache.hadoop.util.StringUtils;
  16. import java.io.IOException;
  17. import java.util.Map;
  18. import java.util.Set;
  19. import java.util.TreeMap;
  21. public class OneStepFlowSumSort {
  23.     public static class OneStepFlowSumMapper extends Mapper<LongWritable, Text, Text, FlowBean> {
  25.         Text k = new Text();
  26.         FlowBean v = new FlowBean();
  28.         @Override
  29.         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  30.             //将读取到的每一行数据进行字段的切分
  31.             String line = value.toString();
  32.             String[] fields = StringUtils.split(line, ' ');
  34.             //抽取我们业务所需要的字段
  35.             String phoneNum = fields[1];
  36.             long upFlow = Long.parseLong(fields[fields.length - 3]);
  37.             long downFlow = Long.parseLong(fields[fields.length - 2]);
  39.             k.set(phoneNum);
  40.             v.set(upFlow, downFlow);
  42.             context.write(k, v);
  44.         }
  45.     }
  47.     public static class OneStepFlowSumReducer extends Reducer<Text, FlowBean, Text, FlowBean> {
  48.         //在这里进行reduce端的局部缓存TreeMap
  49.         TreeMap<FlowBean,Text> treeMap = new TreeMap<FlowBean, Text>();
  51.         //这里reduce方法接收到的key就是某一组《a手机号，bean》《a手机号，bean》   《b手机号，bean》《b手机号，bean》当中的第一个手机号
  52.         //这里reduce方法接收到的values就是这一组kv对中的所以bean的一个迭代器
  53.         @Override
  54.         protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {
  55.             long upFlowCount = 0;
  56.             long downFlowCount = 0;
  58.             for(FlowBean bean : values){
  59.                 upFlowCount += bean.getUpFlow();
  60.                 downFlowCount += bean.getDownFlow();
  61.             }
  63.             FlowBean sumbean = new FlowBean();
  64.             sumbean.set(upFlowCount, downFlowCount);
  66.             Text text = new Text(key.toString());
  68.             treeMap.put(sumbean, text);
  69.         }
  71.         //这里进行我们全局的最终输出
  72.         @Override
  73.         protected void cleanup(Context context) throws IOException, InterruptedException {
  74.             Set<Map.Entry<FlowBean,Text>> entrySet = treeMap.entrySet();
  75.             for(Map.Entry<FlowBean,Text> ent :entrySet){
  76.                 context.write(ent.getValue(), ent.getKey());
  77.             }
  78.         }
  80.         public static void main(String[] args) throws Exception {
  81.             Configuration conf = new Configuration();
  82.             Job job = Job.getInstance(conf);
  84.             job.setJarByClass(OneStepFlowSumSort.class);
  86.             //告诉程序，我们的程序所用的mapper类和reducer类是什么
  87.             job.setMapperClass(OneStepFlowSumMapper.class);
  88.             job.setReducerClass(OneStepFlowSumReducer.class);
  90.             //告诉框架，我们程序输出的数据类型
  91.             job.setMapOutputKeyClass(Text.class);
  92.             job.setMapOutputValueClass(FlowBean.class);
  94.             job.setOutputKeyClass(Text.class);
  95.             job.setOutputValueClass(FlowBean.class);
  97.             //告诉框架，我们程序使用的数据读取组件 结果输出所用的组件是什么
  98.             //TextInputFormat是mapreduce程序中内置的一种读取数据组件  准确的说 叫做 读取文本文件的输入组件
  99.             job.setInputFormatClass(TextInputFormat.class);
  100.             job.setOutputFormatClass(TextOutputFormat.class);
  102.             //告诉框架，我们要处理的数据文件在那个路劲下
  103.             FileInputFormat.setInputPaths(job, new Path("/Users/jdxia/Desktop/website/hdfs/flowsum/input/"));
  105.             //告诉框架，我们的处理结果要输出到什么地方
  106.             FileOutputFormat.setOutputPath(job, new Path("/Users/jdxia/Desktop/website/hdfs/flowsum/output/"));
  108.             boolean res = job.waitForCompletion(true);
  110.             System.exit(res?0:1);
  111.         }
  112.     }
  114. }