对不同的手机号码分成不同文件

接着上面的例子

默认的分区规则

image.png

分区类

  1. package com.folwsum;
  3. import org.apache.hadoop.mapreduce.Partitioner;
  5. import java.util.HashMap;
  7. public class ProvivcePartitioner extends Partitioner {
  8.     private static HashMap<String, Integer> provinceMap = new HashMap<String, Integer>();
  10.     static{
  11.         //我们这边就提前设置手机号码对应的分区
  12.         provinceMap.put("135", 0);
  13.         provinceMap.put("136", 1);
  14.         provinceMap.put("137", 2);
  15.         provinceMap.put("138", 3);
  16.         provinceMap.put("139", 4);
  17.     }
  19.     @Override
  20.     public int getPartition(Object o, Object o2, int numPartitions) {
  21.         //根据手机的前3位,进行取他的值,就是上面定义的
  22.         Integer code = provinceMap.get(o.toString().substring(0, 3));
  23.         if(code != null){
  24.             return code;
  25.         }
  26.         //没有取到就分到5去
  27.         return 5;
  28.     }
  29. }

任务类

主要是main方法里面的

  1. package com.folwsum;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.fs.FileSystem;
  5. import org.apache.hadoop.fs.Path;
  6. import org.apache.hadoop.io.LongWritable;
  7. import org.apache.hadoop.io.Text;
  8. import org.apache.hadoop.mapreduce.Job;
  9. import org.apache.hadoop.mapreduce.Mapper;
  10. import org.apache.hadoop.mapreduce.Reducer;
  11. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  12. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  13. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  14. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  15. import org.apache.hadoop.util.StringUtils;
  17. import java.io.IOException;
  19. public class FlowSumProvince {
  21.     public static class ProvinceFlowSumMapper extends Mapper<LongWritable, Text, Text, FlowBean> {
  22.          Text k = new Text();
  23.          FlowBean v = new FlowBean();
  25.         @Override
  26.         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  27.             //将读取到的每一行数据进行字段的切分
  28.             String line = value.toString();
  29.             String[] fields = StringUtils.split(line, ' ');
  31.             //抽取我们业务所需要的字段
  32.             String phoneNum = fields[1];
  33.             long upFlow = Long.parseLong(fields[fields.length - 3]);
  34.             long downFlow = Long.parseLong(fields[fields.length - 2]);
  36.             k.set(phoneNum);
  37.             v.set(upFlow, downFlow);
  39.             context.write(k, v);
  40.         }
  41.     }
  43.     public static class ProvinceFlowSumReducer extends Reducer<Text, FlowBean, Text, FlowBean> {
  45.         @Override
  46.         protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {
  47.             long upFlowCount = 0;
  48.             long downFlowCount = 0;
  50.             for(FlowBean bean : values){
  51.                 upFlowCount += bean.getUpFlow();
  52.                 downFlowCount += bean.getDownFlow();
  53.             }
  55.             FlowBean sumbean = new FlowBean();
  56.             sumbean.set(upFlowCount, downFlowCount);
  57.             context.write(key, sumbean);
  58.         }
  59.     }
  61.     public static void main(String[] args) throws Exception {
  62.         Configuration conf = new Configuration();
  63.         Job job = Job.getInstance(conf);
  65.         job.setJarByClass(FlowSumProvince.class);
  67.         //告诉程序，我们的程序所用的mapper类和reducer类是什么
  68.         job.setMapperClass(ProvinceFlowSumMapper.class);
  69.         job.setReducerClass(ProvinceFlowSumReducer.class);
  71.         //告诉框架，我们程序输出的数据类型
  72.         job.setMapOutputKeyClass(Text.class);
  73.         job.setMapOutputValueClass(FlowBean.class);
  75.         job.setOutputKeyClass(Text.class);
  76.         job.setOutputValueClass(FlowBean.class);
  78.         //设置我们的shuffer的分区组件使用我们自定义的组件
  79.         job.setPartitionerClass(ProvivcePartitioner.class);
  81.         //这里设置我们的reduce task个数  默认是一个partition分区对应一个reduce task 输出文件也是一对一
  82.         //如果我们的Reduce task个数 < partition分区数  就会报错Illegal partition
  83.         //如果我们的Reduce task个数 > partition分区数 不会报错，会有空文件产生
  84.         //如果我们的Reduce task个数  = 1  partitoner组件就无效了  不存在分区的结果
  86.         //这边设置为6,因为没有匹配到的就到第5个
  87.         job.setNumReduceTasks(6);
  89.         //告诉框架，我们程序使用的数据读取组件 结果输出所用的组件是什么
  90.         //TextInputFormat是mapreduce程序中内置的一种读取数据组件  准确的说 叫做 读取文本文件的输入组件
  91.         job.setInputFormatClass(TextInputFormat.class);
  92.         job.setOutputFormatClass(TextOutputFormat.class);
  94.         //告诉框架，我们要处理的数据文件在那个路劲下
  95.         FileInputFormat.setInputPaths(job, new Path("/Users/jdxia/Desktop/website/hdfs/flowsum/input"));
  96.         //如果有这个文件夹就删除
  97.         Path out = new Path("/Users/jdxia/Desktop/website/hdfs/flowsum/output");
  98.         FileSystem fileSystem = FileSystem.get(conf);
  99.         if (fileSystem.exists(out)) {
  100.             fileSystem.delete(out, true);
  101.         }
  102.         //告诉框架，我们的处理结果要输出到什么地方
  103.         FileOutputFormat.setOutputPath(job, out);
  105.         boolean res = job.waitForCompletion(true);
  107.         System.exit(res?0:1);
  108.     }
  109. }