07 MapREduce分区与自定义数据

  1. package com.etc;
  3. import java.io.DataInput;
  4. import java.io.DataOutput;
  5. import java.io.IOException;
  6. import java.net.URI;
  7. import java.net.URISyntaxException;
  9. import org.apache.hadoop.conf.Configuration;
  10. import org.apache.hadoop.fs.FileSystem;
  11. import org.apache.hadoop.fs.Path;
  12. import org.apache.hadoop.io.IntWritable;
  13. import org.apache.hadoop.io.LongWritable;
  14. import org.apache.hadoop.io.NullWritable;
  15. import org.apache.hadoop.io.Text;
  16. import org.apache.hadoop.io.WritableComparable;
  17. import org.apache.hadoop.io.compress.CompressionCodec;
  18. import org.apache.hadoop.io.compress.GzipCodec;
  19. import org.apache.hadoop.mapreduce.Job;
  20. import org.apache.hadoop.mapreduce.Mapper;
  21. import org.apache.hadoop.mapreduce.Partitioner;
  22. import org.apache.hadoop.mapreduce.Reducer;
  23. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  24. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  25. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  27. public class RectangleSort {
  29.     static final String INPUT_PATH = "hdfs://hadoop0:9000/data/Rectangle.txt";
  30.     static final String OUTPUT_PATH = "hdfs://hadoop0:9000/output/ide";
  32.     public static void main(String[] args) throws Throwable, URISyntaxException {
  33.         Configuration conf = new Configuration();
  34.         // conf.setBoolean("mapred.compress.map.output", true);
  35.         // conf.setClass("mapred.map.output. compression.codec",
  36.         // GzipCodec.class,CompressionCodec.class);
  38.         FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), conf);
  39.         Path outpath = new Path(OUTPUT_PATH);
  40.         if (fileSystem.exists(outpath)) {
  41.             fileSystem.delete(outpath, true);
  42.         }
  43.         Job job = new Job(conf, "RectangleSort");
  44.         job.setJarByClass(RectangleWritable.class);
  45.         org.apache.hadoop.mapreduce.lib.input.FileInputFormat.setInputPaths(job, INPUT_PATH);
  47.         job.setInputFormatClass(TextInputFormat.class);
  48.         job.setMapperClass(MyMapper.class);
  49.         job.setMapOutputKeyClass(RectangleWritable.class);
  50.         job.setMapOutputValueClass(NullWritable.class);
  51.         job.setReducerClass(MyReducer.class);
  52.         job.setOutputKeyClass(IntWritable.class);
  53.         job.setOutputValueClass(IntWritable.class);
  54.         FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH));
  55.         // FileOutputFormat.setCompressOutput(job, true); // job使用压缩
  56.         // FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class); // 设置压缩格式
  57.         job.setOutputFormatClass(TextOutputFormat.class);
  58.         job.setPartitionerClass(MyPatitioner.class);
  59.         job.setNumReduceTasks(2);
  60.         job.waitForCompletion(true);
  61.     }
  63.     /**
  64.      * Map函数
  65.      * @author root
  66.      *
  67.      */
  68.     static class MyMapper extends Mapper<LongWritable, Text, RectangleWritable, NullWritable> {
  70.         protected void map(LongWritable k1, Text v1, Context context) throws IOException, InterruptedException {
  71.             String[] splites = v1.toString().split(" ");
  72.             RectangleWritable k2 = new RectangleWritable(Integer.parseInt(splites[0]), Integer.parseInt(splites[1]));
  73.             context.write(k2, NullWritable.get());
  74.         };
  75.     }
  77.     /**
  78.      * Reduce函数
  79.      * @author root
  80.      *
  81.      */
  82.     static class MyReducer extends Reducer<RectangleWritable, NullWritable, IntWritable, IntWritable> {
  83.         protected void reduce(RectangleWritable k2, Iterable<NullWritable> v2s, Context context)
  84.                 throws IOException, InterruptedException {
  85.             context.write(new IntWritable(k2.getLength()), new IntWritable(k2.getWidth()));
  86.         };
  87.     }
  89. }
  91. class RectangleWritable implements WritableComparable {
  93.     // 矩形长和宽
  94.     int length, width;
  96.     public int getLength() {
  97.         return length;
  98.     }
  100.     public void setLength(int length) {
  101.         this.length = length;
  102.     }
  104.     public int getWidth() {
  105.         return width;
  106.     }
  108.     public void setWidth(int width) {
  109.         this.width = width;
  110.     }
  112.     // 无参构造方法
  113.     public RectangleWritable() {
  114.         super();
  115.     }
  117.     // 有参构造
  118.     public RectangleWritable(int length, int width) {
  119.         super();
  120.         this.length = length;
  121.         this.width = width;
  122.     }
  124.     // 序列化写
  125.     public void write(DataOutput out) throws IOException {
  126.         out.writeInt(length);
  127.         out.writeInt(width);
  128.     }
  130.     // 序列化读
  131.     public void readFields(DataInput in) throws IOException {
  132.         this.length = in.readInt();
  133.         this.width = in.readInt();
  134.     }
  136.     // 比较方法，实现按面积大小排序
  137.     public int compareTo(Object o) {
  138.         RectangleWritable to = (RectangleWritable) o;
  139.         if (this.getLength() * this.getWidth() > to.getLength() * to.getWidth())
  140.             return 1;
  141.         if (this.getLength() * this.getWidth() < to.getLength() * to.getWidth())
  142.             return -1;
  143.         return 0;
  144.     }
  146. }
  148. /**
  149.  * 判断一个矩形是否三正方形，并交给特定的Reduce人物处理的分区规则
  150.  * 
  151.  * @author root
  152.  *
  153.  */
  154. class MyPatitioner extends Partitioner<RectangleWritable, NullWritable> {
  155.     @Override
  156.     public int getPartition(RectangleWritable k2, NullWritable v2, int numReduceTasks) {
  158.         if (k2.getLength() == k2.getWidth()) {
  159.             // 正方形在任务0中汇总
  160.             return 0;
  161.         } else
  162.             // 长方形在任务1中汇总
  163.             return 1;
  164.     }
  165. }