MapReduce的自定义分区与ReduceTask数量

        本篇博客小菌为大家带来的是MapReduce的自定义分区与ReduceTask内容的分享(ReduceMap具体计算流程见《MapReduce中shuffle阶段概述及计算任务流程》)。

        在MapReduce中。通过指定分区，会将同一个分区的数据发送到同一个reduce中，例如为了数据的统计，可以把一批类似的数据发 送到同一个reduce当中去，在同一个reduce中统计相同类型的数据，就可以实现类似数据的分区，统计等。
        直观的说就是相同类型的数据，送到一起去处理，在reduce当中默认分区只有1个。

        MapReduce当中的分区类图

        假设现在有个业务需求,有一个文本文件partition.txt。其中第六个字段表示开奖结果数值，现在以15为分界点，将15以上的结果保存到一个文件，15以下的结果保存到一个文件。

        那我们应该怎么做呢?
        按照正常的思维,我们应该先创建两个目录,在Map中拿到数据后,把开奖结果数值作为key,整行数据作为value传递给reduce,然后reduce再根据key值大小做出一个判断,把list数据存入到不同的目录中…

        虽然步骤没有错,但总归有些繁琐。事实上,MapReduce中的分区机制就可以很好的解决这个问题!

        先让我们来看下MapReduce自带的默认分区算法:

        对key 进行哈希，获取到一个哈希值，用这个哈希值与reducetask的数量取余。余几，这个数据就放在余数编号的partition中。

        但很明显上面提到的问题最终是根据数值大小进行分区,所以这个分区算法并不适用,所以我们需要自定义分区!

        自定义分区的步骤见下:

第一步：定义mapper

        这里的mapper程序不做任何逻辑，也不对key，与value做任何改变，只是接收数据，然后往下发送。

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
public class MyMapper extends Mapper<LongWritable,Text,Text, NullWritable>{
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        context.write(value,NullWritable.get());
    }
}

  
 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
 

第二步：定义reducer逻辑

        reducer也不做任何处理，将数据原封不动的输出即可。

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;

public class MyReducer extends Reducer<Text, NullWritable,Text,NullWritable> {
    @Override
    protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
        context.write(key,NullWritable.get());
    }
}

  
 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
 

第三步：自定义partitioner(分区)

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
/**
 * 这里的输入类型与map阶段的输出类型相同
 */
public class MyPartitioner extends Partitioner<Text,NullWritable> {
    /**
     * 返回值表示数据要去到哪个分区
     * 返回值只是一个分区的标记，标记所有相同的数据去到指定的分区
     */
    @Override
    public int getPartition(Text text, NullWritable nullWritable, int i) {
        String result = text.toString().split("\t")[5];
        System.out.println(result);
        if (Integer.parseInt(result) > 15){
            return 1;
        }else{
            return 0;
        }
    }
}

  
 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
 

第四步：程序main函数入口

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
public class PartitionMain  extends Configured implements Tool {
    public static void main(String[] args) throws  Exception{
        int run = ToolRunner.run(new Configuration(), new PartitionMain(), args);
        System.exit(run);
    }
    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(super.getConf(), PartitionMain.class.getSimpleName());
        job.setJarByClass(PartitionMain.class);
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        TextInputFormat.addInputPath(job,new Path("hdfs://192.168.100.129:8020/partitioner"));
        TextOutputFormat.setOutputPath(job,new Path("hdfs://192.168.100.129:8020/outpartition"));
        job.setMapperClass(MyMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);
        job.setReducerClass(MyReducer.class);
        /**
         * 设置分区类，以及reducetask的个数，注意reduceTask的个数一定要与
         * 分区数保持一致
         */
         //使用combine
        job.setPartitionerClass(MyPartitioner.class);
        // 设置reduceTask的个数
        job.setNumReduceTasks(2);
        boolean b = job.waitForCompletion(true);
        return b?0:1;
    }
}

  
 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
  
23
  
24
  
25
  
26
  
27
  
28
  
29
  
30
  
31
  
32
  
33
  
34
  
35
  
36
  
37
  
38
  
39
  
40
  
41
  
42
  
43
 

        注意：分区的案例，只能打成jar包发布到集群上面去运行，本地模式已经不能正常运行了

运行结果


        可以发现,数值大于15的放在了一个分区,小于等于15的又放在了另一个分区,说明我们自定义分区成功了!

        在这个过程中最最重要的一点小菌有必要再重复一遍,在设置分区类，以及reducetask的个数，注意reduceTask的个数一定要与分区数保持一致,否则分区将不具有任何意义!

        本次的分享就到这里了,受益的小伙伴们记得点赞关注小菌~小菌后续会更新更多hadoop的相关内容,敬请关注(^_−)☆

文章来源: alice.blog.csdn.net，作者：大数据梦想家，版权归原作者所有，如需转载，请联系作者。

原文链接：alice.blog.csdn.net/article/details/103074146