首页 > 其他 > 详细

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

时间：2016-06-01 02:21:43      阅读：276      评论：0      收藏：0      [点我收藏+]
上一个博文中，我们分别介绍了使用combiner 和in-map 聚合 来优化 map reduce job。
这一篇中，继续优化这个统计字符数量的mr 程序。
再次介绍下map reduce程序的目标： 统计输入文件中 a~z 字符分别有多少个。 A 与 a 都统计到 a 中。 
先看一下上次运行的Job Counters 计数，可以看到Reducer Task的个数为1.

点击(此处)折叠或打开

  1.         Job Counters 
  2.                 Launched map tasks=3
  3.                 Launched reduce tasks=1
  4.                 Data-local map tasks=3
  5.                 Total time spent by all maps in occupied slots (ms)=1207744
  6.                 Total time spent by all reduces in occupied slots (ms)=341424
  7.     
知识点：MR Job 默认的reducer 个数为1.  reducer的个数可以通过job.setNumReducerTasks(num) 来进行设置。
留个问题：那map 任务的个数怎么设置？

假如数据量很大时， 一个reducer task 可能成为任务的bottle neck。 那我们手工设置一下reducer 个数。

点击(此处)折叠或打开

  1. @Override
  2.         public int run(String[] args) throws Exception {
  3.                 //valid the parameters
  4.                 if(args.length !=2){
  5.                         return -1;
  6.                 }

  8.                 Job job = Job.getInstance(getConf(), "MyWordCountJob");
  9.                 job.setJarByClass(MyWordCountJob.class);

  11.                 Path inPath = new Path(args[0]);
  12.                 Path outPath = new Path(args[1]);

  14.                 outPath.getFileSystem(getConf()).delete(outPath,true);
  15.                 TextInputFormat.setInputPaths(job, inPath);
  16.                 TextOutputFormat.setOutputPath(job, outPath);


  19.                 job.setMapperClass(MyWordCountJob.MyWordCountMapper.class);
  20.                 job.setReducerClass(MyWordCountJob.MyWordCountReducer.class);
  21.                 job.setInputFormatClass(TextInputFormat.class);
  22.                 job.setOutputFormatClass(TextOutputFormat.class);

  24.                 job.setMapOutputKeyClass(Text.class);
  25.                 job.setMapOutputValueClass(IntWritable.class);
  26.                 job.setOutputKeyClass(Text.class);
  27.                 job.setOutputValueClass(IntWritable.class);

  29.                 job.setNumReduceTasks(3);
  30.                 return job.waitForCompletion(true)?0:1;
  31.         }
编译后再次运行查看日志 可以看到 reducer的个数是3，符合我们的预期。但问题还没有完！

点击(此处)折叠或打开

  1. File System Counters
  2.                 FILE: Number of bytes read=642
  3.                 FILE: Number of bytes written=507033
  4.                 FILE: Number of read operations=0
  5.                 FILE: Number of large read operations=0
  6.                 FILE: Number of write operations=0
  7.                 HDFS: Number of bytes read=556
  8.                 HDFS: Number of bytes written=107
  9.                 HDFS: Number of read operations=18
  10.                 HDFS: Number of large read operations=0
  11.                 HDFS: Number of write operations=6
  12.         Job Counters
  13.                 Launched map tasks=3
  14.                 Launched reduce tasks=3
  15.                 Data-local map tasks=3
  16.                 Total time spent by all maps in occupied slots (ms)=1207744
  17.                 Total time spent by all reduces in occupied slots (ms)=341424
  18.         Map-Reduce Framework
  19.                 Map input records=8
  20.                 Map output records=78
  21.                 Map output bytes=468
  22.                 Map output materialized bytes=678
  23.                 Input split bytes=399
  24.                 Combine input records=0
  25.                 Combine output records=0
  26.                 Reduce input groups=26
  27.                 Reduce shuffle bytes=678
  28.                 Reduce input records=78
  29.                 Reduce output records=26
  30.                 Spilled Records=156
  31.                 Shuffled Maps =9
  32.                 Failed Shuffles=0
  33.                 Merged Map outputs=9
  34.                 GC time elapsed (ms)=507
  35.                 CPU time spent (ms)=7770
  36.                 Physical memory (bytes) snapshot=1329672192
  37.                 Virtual memory (bytes) snapshot=5978918912
  38.                 Total committed heap usage (bytes)=1004273664

下面看一下输出，有三个输出文件，正常，一个reducer 对应一个输出文件。

点击(此处)折叠或打开

  1. [train@sandbox MyWordCount]$ hdfs dfs -ls output
  2. Found 4 items
  3. -rw-r--r-- 3 train hdfs 0 2016-05-11 11:48 output/_SUCCESS
  4. -rw-r--r-- 3 train hdfs 37 2016-05-11 11:48 output/part-r-00000
  5. -rw-r--r-- 3 train hdfs 34 2016-05-11 11:48 output/part-r-00001
  6. -rw-r--r-- 3 train hdfs 36 2016-05-11 11:48 output/part-r-00002

我们看一下结果文件中的内容

点击(此处)折叠或打开

  1. [train@sandbox MyWordCount]$ hdfs dfs -cat output/part-r-00000
  2. b 3
  3. e 11
  4. h 8
  5. k 3
  6. n 4
  7. q 3
  8. t 4
  9. w 7
  10. z 3
  11. [train@sandbox MyWordCount]$ hdfs dfs -cat output/part-r-00001
  12. c 4
  13. f 4
  14. i 5
  15. l 6
  16. o 12
  17. r 13
  18. u 6
  19. x 3
  20. [train@sandbox MyWordCount]$ hdfs dfs -cat output/part-r-00002
  21. a 8
  22. d 4
  23. g 3
  24. j 3
  25. m 7
  26. p 3
  27. s 4
  28. v 0
  29. y 6
发现三个文件中没有按顺序输出，现在我们想要的是a~h 放到一个文件中，i~q 放到第二个文件中、r~z 放到第三个文件中。我们应该怎么办？
这就要用到我们本次将要介绍的利器： Partitioner 。
Partition 就是用来决定map的输出会输送到哪儿个reducer中。当只有一个reducer 时，不会调用Partitioner，即使配置了也不会调用。
hadoop 框架中默认的Partitioner 是HashPartitioner。 它是根据key的hash值对reducer个数取余进行分配的。
说明一下：返回0即本记录将发往第一个reducer,返回1则本记录发往第二个reducer.依次类推。

点击(此处)折叠或打开

  1. public class HashPartitioner<K, V> extends Partitioner<K, V> {
  2.  
  3.   public int getPartition(K key, V value,
  4.                           int numReduceTasks) {
  5.     return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
  6.   }
  7. }
那接下来我们要定义自己的一个Partitioner ,它要按照我们预定的a~h 发往第一个reducer，i~q 发往第二个reducer、r~z发往第三个reducer。
 

点击(此处)折叠或打开

  1. public static class MyWordCountPartitioner extends Partitioner<Text,IntWritable>{

  3.                 @Override
  4.                 public int getPartition(Text key, IntWritable value, int numPartitions) {
  5.                         char c =key.toString().charAt(0);
  6.                         if(c>=‘a‘&& c<‘i‘)
  7.                                 return 0;
  8.                         else if(c>=‘i‘ && c<‘q‘)
  9.                                 return 1;
  10.                         else
  11.                                 return 2;
  12.                 }

  14.         }
运行查看结果，可见实现了我们的预期。

点击(此处)折叠或打开

  1. [train@sandbox MyWordCount]$ hdfs dfs -cat output/part-r-00000
  2. a 8
  3. b 3
  4. c 4
  5. d 4
  6. e 11
  7. f 4
  8. g 3
  9. h 8
  10. [train@sandbox MyWordCount]$ hdfs dfs -cat output/part-r-00001
  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 MyWordCount]$ hdfs dfs -cat output/part-r-00002
  20. q 3
  21. r 13
  22. s 4
  23. t 4
  24. u 6
  25. v 0
  26. w 7
  27. x 3
  28. y 6
  29. z 3
下面照例把整个代码贴上

点击(此处)折叠或打开

  1. package wordcount;

  3. import java.io.IOException;
  4. import java.util.HashMap;
  5. import java.util.Map;

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

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

  26.         public static class MyWordCountMapper extends
  27.                         Mapper<LongWritable, Text, Text, IntWritable> {
  28.                 Logger log = Logger.getLogger(MyWordCountJob.class);
  29.                 Map<Character,Integer> map = new HashMap<Character,Integer>();

  31.                 Text mapKey = new Text();
  32.                 IntWritable mapValue = new IntWritable(1);
  33.                 @Override
  34.                 protected void map(LongWritable key, Text value, Context context)
  35.                                 throws IOException, InterruptedException {
  36.                         for(char c :value.toString().toLowerCase().toCharArray()){
  37.                                 if(c>=‘a‘ && c <=‘z‘){
  38.                                         map.put(c,map.get(c)+1);
  39.                                 }
  40.                         }
  41.                 }

  43.                 @Override
  44.                 protected void cleanup(Context context) throws IOException,
  45.                                 InterruptedException {
  46.                         for(char key : map.keySet()){
  47.                                 mapKey.set(String.valueOf(key));
  48.                                 mapValue.set(map.get(key));
  49.                                 context.write(mapKey, mapValue);
  50.                         }

  52.                 }

  54.                 @Override
  55.                 protected void setup(Context context) throws IOException,
  56.                                 InterruptedException {
  57.                         for(char c=‘a‘;c<=‘z‘ ;c++){
  58.                                 map.put(c, 0);
  59.                         }
  60.                 }

  62.         }

  64.         public class MyWordCountPartitioner extends Partitioner<Text,IntWritable>{

  66.                 @Override
  67.                 public int getPartition(Text key, IntWritable value, int numPartitions) {
  68.                         char c =key.toString().charAt(0);
  69.                         if(c>=‘a‘&& c<‘i‘)
  70.                                 return 0;
  71.                         else if(c>=‘i‘ && c<‘q‘)
  72.                                 return 1;
  73.                         else
  74.                                 return 2;
  75.                 }

  77.         }

  79.         public static class MyWordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
  80.                 Text rkey = new Text();
  81.                 IntWritable rvalue = new IntWritable(1);
  82.                 @Override
  83.                 protected void reduce(Text key, Iterable<IntWritable> values,Context context)
  84.                                 throws IOException, InterruptedException {
  85.                         int n=0;
  86.                         for(IntWritable value :values){
  87.                                 n+= value.get();
  88.                         }
  89.                         rvalue.set(n);
  90.                         context.write(key, rvalue);
  91.                 }
  92.         }

  94.         @Override
  95.         public int run(String[] args) throws Exception {
  96.                 //valid the parameters
  97.                 if(args.length !=2){
  98.                         return -1;
  99.                 }

  101.                 Job job = Job.getInstance(getConf(), "MyWordCountJob");
  102.                 job.setJarByClass(MyWordCountJob.class);

  104.                 Path inPath = new Path(args[0]);
  105.                 Path outPath = new Path(args[1]);

  107.                 outPath.getFileSystem(getConf()).delete(outPath,true);
  108.                 TextInputFormat.setInputPaths(job, inPath);
  109.                 TextOutputFormat.setOutputPath(job, outPath);


  112.                 job.setMapperClass(MyWordCountJob.MyWordCountMapper.class);
  113.                 job.setReducerClass(MyWordCountJob.MyWordCountReducer.class);
  114.                 job.setPartitionerClass(MyWordCountPartitioner.class);
  115.                 job.setInputFormatClass(TextInputFormat.class);
  116.                 job.setOutputFormatClass(TextOutputFormat.class);
  117.                 job.setMapOutputKeyClass(Text.class);
  118.                 job.setMapOutputValueClass(IntWritable.class);
  119.                 job.setOutputKeyClass(Text.class);
  120.                 job.setOutputValueClass(IntWritable.class);

  122.                 job.setNumReduceTasks(3);
  123.                 return job.waitForCompletion(true)?0:1;
  124.         }
  125.         public static void main(String [] args){
  126.                 int result = 0;
  127.                 try {
  128.                         result = ToolRunner.run(new Configuration(), new MyWordCountJob(), args);
  129.                 } catch (Exception e) {
  130.                         e.printStackTrace();
  131.                 }
  132.                 System.exit(result);
  133.         }

  135. }
最后再唠叨两句： 一般情况要，自定义partitioner 是为了解决数据分布不均的情况，又叫数据倾斜。 而且自定义的partitioner 要保证，相同的key要发往相同的reducer。

 

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

原文：http://blog.itpub.net/30066956/viewspace-2108214/
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
分享档案
更多>
2021年09月23日 (328)
2021年09月24日 (313)
2021年09月17日 (191)
2021年09月15日 (369)
2021年09月16日 (411)
2021年09月13日 (439)
2021年09月11日 (398)
2021年09月12日 (393)
2021年09月10日 (160)
2021年09月08日 (222)
最新文章
更多>
教程昨日排行
更多>
友情链接
汇智网   PHP教程   插件网  
关于我们 - 联系我们 - 留言反馈 - 联系我们:wmxa8@hotmail.com
© 2014 bubuko.com 版权所有
打开技术之扣，分享程序人生！