在上一篇我们的一个案例中，我们对出现单词的各个统计进行了mapreduce的入门学习，其中重要的两个步骤就是对map阶段和reduce阶段进行了代码的编写。但是在shuffle阶段使用了默认的分区、排序、规约、分组。在我们工作当中应该更多的是将不同的分区交给不同的reduce，从而产生不同的结果。
在MapReduce中，通过我们指定的分区，会将同一个分区的数据发送到同一个reduce当中去处理。例如：为了数据的统计，可以把一批类似的数据发送到同一个reduce当中，在同一个reduce当中统计相同类型的数据，就可以实现类似的数据分区和统计等。其实就死相同的类型数据，有共性的数据送到一起出来，reduce当中默认的分区只有一个。

单词统计任务拓展

还是跟上篇案例类似，假设有个a.txt文件300M，在单词统计的基础上要求对单词长度>=5放到一个结果集，<5放到另外一个结果集。此时的reduceTask就不是下图这个了，而是存在两个reduceTask，也就是说同一个分区的数据会放到同一个reduce，同一个reduce会产生一个文件
image.pngimage.png

代码实现

新增PartitionerOwn类

要实现分区必须要重写Partitioner，然后自定义我们的分区规则

  1. package com.dongnaoedu.network.hadoop.mapreduce.partition;
  3. import org.apache.hadoop.io.LongWritable;
  4. import org.apache.hadoop.io.Text;
  5. import org.apache.hadoop.mapreduce.Partitioner;
  8. public class PartitionerOwn extends Partitioner<Text, LongWritable>{
  9.     /**
  10.      * text：表示k2
  11.      * longWritable:表示v2
  12.      * i:reduce个数
  13.      */
  15.     @Override
  16.     public int getPartition(Text text,LongWritable longWritable,int i) {
  17.         // 如果单词的长度>=5进入第一个分区==>第一个reduceTask==>reduce编号是0
  18.        if(text.toString().length() >= 5){
  19.            return 0;
  20.        }else{
  21.            // 如果长度<5，进入第二个分区==>第二个reduceTask==>reduce编号是1
  22.            return 1;
  23.        }
  24.     }
  25. }

主方法设置分区代码

  1. package com.dongnaoedu.network.hadoop.mapreduce.partition;
  3. import com.dongnaoedu.network.hadoop.mapreduce.WordCountMapper;
  4. import com.dongnaoedu.network.hadoop.mapreduce.WordCountReduce;
  5. import org.apache.hadoop.conf.Configuration;
  6. import org.apache.hadoop.conf.Configured;
  7. import org.apache.hadoop.fs.Path;
  8. import org.apache.hadoop.io.LongWritable;
  9. import org.apache.hadoop.io.Text;
  10. import org.apache.hadoop.mapreduce.Job;
  11. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  12. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  13. import org.apache.hadoop.util.Tool;
  14. import org.apache.hadoop.util.ToolRunner;
  17. public class JobMain extends Configured implements Tool{
  18.     @Override
  19.     public int run(String[] strings) throws Exception{
  20.         Job job = Job.getInstance(super.getConf(), "mapReduce_wordCount");
  22.         //打包放在集群下运行，需要做一个配置
  23.         job.setJarByClass(JobMain.class);
  25.         // 第一步：设置读取文件的类：k1 v1
  26.         job.setInputFormatClass(TextInputFormat.class);
  27.         TextInputFormat.addInputPath(job, new Path("hdfs://node01:8082/wordcount"));
  29.         // 第二步：设置mapper类
  30.         job.setMapperClass(WordCountMapper.class);
  31.         // 设置map阶段输出类型
  32.         job.setMapOutputKeyClass(Text.class);
  33.         job.setMapOutputValueClass(LongWritable.class);
  35.         // 第三 四 五 六步采用默认方式（分区 排序 规约 分组）
  36.         // 设置我们的分区类
  37.         job.setPartitionerClass(PartitionerOwn.class);
  39.         // 第七步:设置Reduce类
  40.         job.setReducerClass(WordCountReduce.class);
  41.         job.setOutputKeyClass(Text.class);
  42.         job.setOutputValueClass(LongWritable.class);
  44.         // 设置reduce的个数j
  45.         job.setNumReduceTasks(2);
  47.         // 第八步：设置输出类
  48.         job.setOutputFormatClass(TextOutputFormat.class);
  49.         // 设置输出路径
  50.         TextOutputFormat.setOutputPath(job, new Path("hafs://node01:8082/wordcount_cout"));
  52.         boolean b = job.waitForCompletion(true);
  54.         return b ? 0:1;
  55.     }
  57.     public static void main(String[] args) throws Exception{
  58.         Configuration configuration = new Configuration();
  59.         // 启动一个任务  run=0成功  
  60.         int run = ToolRunner.run(configuration, new JobMain(), args);
  61.         System.exit(run);  
  62.     }
  64. }

运行结果

查看我们的hdfs结果，会发现有两个文件，编号为0的就是我们长度>=5（第一个分区），编号为1的就是我们长度<5的（第二个分区）
image.png