知识点连接第一天 如有疑惑移步我的首页

我的首页

补充知识点

2 partition分区 - 图1
Shuffle 主要包括 四个步骤 (分区 排序 规约 分组)

Partition 分区

分区书写步骤

分区流程

2 partition分区 - 图2

创建分区类

  1. public  static class MyPartition extends Partitioner<Text, IntWritable>{
  2.         /**
  3.          * 用于创建 分区规则
  4.          * 继承 org.apache.hadoop.mapreduce.Partitioner;
  5.          * @param text ： map阶段之后 reduce阶段之前的k
  6.          * @param intWritable ： map阶段之后 reduce阶段之前的v
  7.          * @param numPartitions
  8.          * @return 返回int类型 
  9.          */
  10.         /**
  11.         * 重写getPartition方法 
  12.         */
  13.         @Override
  14.         public int getPartition(Text text, IntWritable intWritable, int numPartitions) {
  15.             // 指定规则  例如 如果k是hadoop 那么 返回0 就是0分区 否则 1分区
  16.             if(text.toString().equals("hadoop")){
  17.                 return 0;
  18.             }else{
  19.                 return 1;
  20.             }
  21.         }
  22.     }

让分区类能够被运用 配置

  1. // 这个配置设置谁是分区类 
  2. job.setPartitionerClass(MyPartition.class);
  3. // 设置分区reduce个数  你返回了几个变量分区数就是几  比如我0，1 就是两个分区
  4. job.setNumReduceTasks(2);

完整代码 不同项目

  1. package MapReduce_one;
  4. import org.apache.hadoop.conf.Configuration;
  5. import org.apache.hadoop.fs.Path;
  6. import org.apache.hadoop.io.LongWritable;
  7. import org.apache.hadoop.io.NullWritable;
  8. import org.apache.hadoop.io.Text;
  9. import org.apache.hadoop.mapreduce.Job;
  10. import org.apache.hadoop.mapreduce.Mapper;
  11. import org.apache.hadoop.mapreduce.Partitioner;
  12. import org.apache.hadoop.mapreduce.Reducer;
  13. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  14. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  17. import java.io.IOException;
  19. public class JobMainTwo {
  20.     // main
  21.     public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
  22.         System.setProperty("hadoop.home.dir", "H:\\winutils\\winutils-master\\hadoop-2.7.1");
  23.         Configuration configuration = new Configuration();
  24.         Job job = Job.getInstance(configuration, "text");
  25.         job.setPartitionerClass(JobMainTwo.MyPartition.class);
  26.         // 设置Reduce个数
  27.         job.setReducerClass(JobMainTwo.ReduceP.class);
  28.             job.setOutputKeyClass(Text.class);
  29.             job.setOutputValueClass(NullWritable.class);
  30.         job.setNumReduceTasks(2);
  33.         job.setJarByClass(JobMainTwo.class);
  34.         job.setMapperClass(JobMainTwo.MapP.class);
  35.             job.setMapOutputValueClass(NullWritable.class);
  36.             job.setMapOutputKeyClass(Text.class);
  38.         FileInputFormat.addInputPath(job,new Path("/user/root/baoxian1117.csv"));
  39.         FileOutputFormat.setOutputPath(job,new Path("/user/root/result3"));
  40.         System.exit(job.waitForCompletion(true)?0:1);
  41.     }
  42.     // map
  43.     public static class MapP extends Mapper<LongWritable, Text, Text, NullWritable>{
  44.         @Override
  45.         protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
  46.             context.write(value, NullWritable.get());
  47.         }
  48.     }
  49.     // partition
  50.     public static class MyPartition extends Partitioner<Text, NullWritable>{
  51.         @Override
  52.         public int getPartition(Text text, NullWritable nullWritable, int numPartitions) {
  53.             String[] split = text.toString().split(",");
  54.             if (Integer.parseInt(split[0])>200){
  55.                 return 0;
  56.             }else {
  57.                 return 1;
  58.             }
  59.         }
  60.     }
  61.     // reduce
  62.     public static class ReduceP extends Reducer<Text, NullWritable, Text, NullWritable>{
  63.         @Override
  64.         protected void reduce(Text key, Iterable<NullWritable> values, Reducer<Text, NullWritable, Text, NullWritable>.Context context) throws IOException, InterruptedException {
  65.             context.write(key, NullWritable.get());
  66.         }
  67.     }
  70. }

问题和注意事项

  1. 在本地运行配置得环境可能不支持分区操作 如果可以当然不可以 如果不可以 可以尝试先选择去yarn集群跑一下任务看是否能够正常分区
  2. 注意设置分区数目