ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 大数据 > Hadoop > 一个MapReduce 程序示例 细节决定成败(八)：TotalOrderPartitioner

一个MapReduce 程序示例 细节决定成败(八)：TotalOrderPartitioner

原创 Hadoop 作者：self_control 时间：2016-05-30 16:17:50 0 删除 编辑
在上一个实验中 一个MapReduce 程序示例 细节决定成败(七) ：自定义Key 及RecordReader，使用自定义的RecordReader 从split 中读取信息，封装到自定义的key中。
这一步在使用TotalOrderPartitioner 进行词频统计是非常重要的。
下面进入主题：在一个MapReduce 程序示例 细节决定成败(五) ：Partitioner  中，演示了如何使用一下自定义Partitioner。
我们使用了一个自定义的Partitioner，来实现当设置3个reducer时输出字符的排序问题。为了简化问题，我使用在partitioner中直接进行判断。
这里面其实是有两个我们应该注意的问题。
1、在partitioner代码中写死了，只有3个reducer，那如果我们想要把reducer的个数设置成其它值，还需要改代码。
2、partitioner 的一个作用就是解决数据倾斜的问题，但我们上次写的那个partitioner不适用
那接下来我们介绍的 TotalOrderPartitioner 就完美解决了上述两个问题。
本次涉及到的内容有。 
1、生成样本partition file
2、使用使用CacheFile功能把partition file 分发到各个map task节点。
3、配置使用TotalOrderPartitioner.

因为TotaOrderlPartitioner是已经封装到hadoop 系统中了，所以我们只要正确配置就可以了。
partition file ：描述了每个字符应该进入到哪儿个reducer中的文件，功能就是上一篇中我们简化处理的 a~i 分到0， h~q 分到1，r~z 分到2。
partition file 不用我们手动去写，而且手动去写也很难平均分布各个reducer任务的负载。可以使用InputSampler.RandomSampler完成此项工作的。

点击(此处)折叠或打开

  1.  /**
  2.   *  freq:每个key 可以被抽样次数
  3.   *  numSamples:总样本数  
  4.   */
  5. InputSampler.RandomSampler(double freq, int numSamples);

  6.
  7.  /**
  8.   *  freq:每个key 可以被抽样次数
  9.   *  numSamples:总样本数
  10.   *  maxSplitsSampled:定义最多从多少个splits 上进行抽样
  11.   */
  12. InputSampler.RandomSampler(double freq, int numSamples, int maxSplitsSampled);

默认抽样是在所有的split上进行的，当输入文件很大的时候，使用第一个方法进行抽样代价会很大。
当处理文件很大的时候，加上第三个参数maxSplitsSampled 限定最多处理多少个split。

freq 参数： 被抽样到的split中的每一个key 被选中的概率。 numSamples 代表抽样的总样本数。 比如从10000条数据中抽取 10条数，假如freq 设置为0.5，那按概率大约执行到20条的时候，样本数就够了。
但然后程序会继续执行，每个被抽中的样本去随机替换一条已经抽出的10条记录中的一条，一直到最后。


下面贴一下配置使用TotaOrderlPartitioner的代码 
         关注 getPartitionFile和 new URI 的写法。

点击(此处)折叠或打开

  1.                 job.setPartitionerClass(TotalOrderPartitioner.class);
  2.                 job.setNumReduceTasks(3);
  3.                 RandomSampler<MyKey, NullWritable> sampler
  4.                            = new InputSampler.RandomSampler<MyKey,NullWritable>(0.6, 20, 3);
  5.                 InputSampler.writePartitionFile(job, sampler);

  7.                 String partitionFile = TotalOrderPartitioner.getPartitionFile(job.getConfiguration());
  8.                 URI uri = new URI(partitionFile+"#"+TotalOrderPartitioner.DEFAULT_PATH);
  9.                 job.addCacheFile(uri);


注意： 1、RecordReader的key与map输入输出key和reduce的输入key  要保持一致。  因为取样是从job 配置的InputFormat中得到RecordReader 然后不断的得到Key ,从中取样。 

               取到的样使用TotalOrderPartitioner 得到的partitionFile 是供分配到不同reducer的，所以RecordReader的key 要与Reduer的输入key 一致，又 RecordReader的key value 是输入到 map 中的，

                Reduce 的输入又是Map的输出。绕了半天结论就是： RecordReader的key, Map 的输入输出key ,Reduce的输入key 是要保持一致的。

            2、使用TotalOrderPartitioner前一定要配置了reduceTask 的个数、InputFormat、和上面讨论的 key class。

贴一下结果：


点击(此处)折叠或打开

  1. wordcount.MyKey@212cb585 8
  2. wordcount.MyKey@212cb585 3
  3. wordcount.MyKey@212cb585 4
  4. wordcount.MyKey@212cb585 4
  5. wordcount.MyKey@212cb585 11
  6. wordcount.MyKey@212cb585 4
  7. wordcount.MyKey@212cb585 3
  8. wordcount.MyKey@168497f6 8
  9. wordcount.MyKey@168497f6 5
  10. wordcount.MyKey@168497f6 3
  11. wordcount.MyKey@168497f6 3
  12. wordcount.MyKey@168497f6 6
  13. wordcount.MyKey@168497f6 7
  14. wordcount.MyKey@168497f6 4
  15. wordcount.MyKey@168497f6 12
  16. wordcount.MyKey@168497f6 3
  17. wordcount.MyKey@168497f6 3
  18. wordcount.MyKey@2e67cd84 13
  19. wordcount.MyKey@2e67cd84 4
  20. wordcount.MyKey@2e67cd84 4
  21. wordcount.MyKey@2e67cd84 6
  22. wordcount.MyKey@2e67cd84 7
  23. wordcount.MyKey@2e67cd84 3
  24. wordcount.MyKey@2e67cd84 6
  25. wordcount.MyKey@2e67cd84 3


       注意到key的格式，这是由于我们的MyKey没有覆写toString 方法，输出的时候直接使用了Object的toString实现。
   

点击(此处)折叠或打开

  1. @Override
  2. public String toString(){
  3.        return String.valueOf(this.c);
  4. }
结果分析：  下面数据是运行结果，可以看到抽样划分的界限是a~e  f~p q~z 。

点击(此处)折叠或打开

  1. [train@sandbox src]$ hdfs dfs -cat output/*0
  2. a 8
  3. b 3
  4. c 4
  5. d 4
  6. e 11
  7. [train@sandbox src]$ hdfs dfs -cat output/*1
  8. f 4
  9. g 3
  10. h 8
  11. i 5
  12. j 3
  13. k 3
  14. l 6
  15. m 7
  16. n 4
  17. o 12
  18. p 3
  19. [train@sandbox src]$ hdfs dfs -cat output/*2
  20. q 3
  21. r 13
  22. s 4
  23. t 4
  24. u 6
  25. w 7
  26. x 3
  27. y 6
  28. z 3



下面贴出主程序代码

点击(此处)折叠或打开

  1. package wordcount;

  3. import java.io.IOException;
  4. import java.net.URI;

  6. import org.apache.hadoop.conf.Configuration;
  7. import org.apache.hadoop.conf.Configured;
  8. import org.apache.hadoop.fs.Path;
  9. import org.apache.hadoop.io.IntWritable;
  10. import org.apache.hadoop.io.NullWritable;
  11. import org.apache.hadoop.io.Text;
  12. import org.apache.hadoop.mapred.lib.TotalOrderPartitioner;
  13. import org.apache.hadoop.mapreduce.Job;
  14. import org.apache.hadoop.mapreduce.Mapper;
  15. import org.apache.hadoop.mapreduce.Reducer;
  16. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  17. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  18. import org.apache.hadoop.mapreduce.lib.partition.InputSampler;
  19. import org.apache.hadoop.mapreduce.lib.partition.InputSampler.RandomSampler;
  20. import org.apache.hadoop.util.Tool;
  21. import org.apache.hadoop.util.ToolRunner;
  22. import org.apache.log4j.Logger;

  24. public class MyWordCountJob extends Configured implements Tool {
  25.         Logger log = Logger.getLogger(MyWordCountJob.class);

  27.         public static class MyWordCountMapper extends
  28.                         Mapper<MyKey, NullWritable, MyKey, IntWritable> {
  29.                 Text mKey = new Text();
  30.                 IntWritable mValue = new IntWritable(1);
  31.                 @Override
  32.                 protected void map(MyKey key, NullWritable value, Context context)
  33.                                 throws IOException, InterruptedException {
  34.                         context.write(key, mValue);
  35.                 }
  36.         }

  38.         public static class MyWordCountReducer extends Reducer<MyKey, IntWritable, MyKey, IntWritable> {
  39.                 Text rkey = new Text();
  40.                 IntWritable rvalue = new IntWritable(1);
  41.                 @Override
  42.                 protected void reduce(MyKey key, Iterable<IntWritable> values,Context context)
  43.                                 throws IOException, InterruptedException {

  45.                         int n=0;
  46.                         for(IntWritable value :values){
  47.                                 n+= value.get();
  48.                         }
  49.                         rvalue.set(n);
  50.                         context.write(key, rvalue);
  51.                 }
  52.         }

  54.         @Override
  55.         public int run(String[] args) throws Exception {
  56.                 //valid the parameters
  57.                 if(args.length !=2){
  58.                         return -1;
  59.                 }

  61.                 Job job = Job.getInstance(getConf(), "MyWordCountJob");
  62.                 job.setJarByClass(MyWordCountJob.class);

  64.                 Path inPath = new Path(args[0]);
  65.                 Path outPath = new Path(args[1]);

  67.                 outPath.getFileSystem(getConf()).delete(outPath,true);
  68.                 TextInputFormat.setInputPaths(job, inPath);
  69.                 TextOutputFormat.setOutputPath(job, outPath);


  72.                 job.setMapperClass(MyWordCountJob.MyWordCountMapper.class);
  73.                 job.setReducerClass(MyWordCountJob.MyWordCountReducer.class);

  75.                 job.setInputFormatClass(MyCombinedFilesInputFormat.class);
  76.                 MyCombinedFilesInputFormat.setMaxInputSplitSize(job, 1024*1024*64);
  77.                 job.setOutputFormatClass(TextOutputFormat.class);
  78.                 job.setMapOutputKeyClass(MyKey.class);
  79.                 job.setMapOutputValueClass(IntWritable.class);
  80.                 job.setOutputKeyClass(MyKey.class);
  81.                 job.setOutputValueClass(IntWritable.class);

  83.                 job.setPartitionerClass(TotalOrderPartitioner.class);
  84.                 job.setNumReduceTasks(3);
  85.                 RandomSampler<MyKey, NullWritable> sampler
  86.                            = new InputSampler.RandomSampler<MyKey,NullWritable>(0.6, 20, 3);
  87.                 InputSampler.writePartitionFile(job, sampler);

  89.                 String partitionFile = TotalOrderPartitioner.getPartitionFile(job.getConfiguration());
  90.                 URI uri = new URI(partitionFile+"#"+TotalOrderPartitioner.DEFAULT_PATH);
  91.                 job.addCacheFile(uri);

  93.                 return job.waitForCompletion(true)?0:1;
  94.         }
  95.         public static void main(String [] args){
  96.                 int result = 0;
  97.                 try {
  98.                         result = ToolRunner.run(new Configuration(), new MyWordCountJob(), args);
  99.                 } catch (Exception e) {
  100.                         e.printStackTrace();
  101.                 }
  102.                 System.exit(result);
  103.         }

  105. }



来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/30066956/viewspace-2109331/ ，如需转载，请注明出处，否则将追究法律责任。

2
0
上一篇： 一个MapReduce 程序示例 细节决定成败(七) ：自定义Key 及RecordReader
下一篇： 一个MapReduce 程序示例 细节决定成败(九)：RawComparator
请登录后发表评论 登录
全部评论
self_control

注册时间：2014-12-16

  • 博文量
    49
  • 访问量
    406892

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593