Hadoop Map/Reduce编程模型实现海量数据处理—数字求和
魏仁言 2010.8.24 Map/Reduce编程模型型的原理是:利用一个输入key/value pair 集合来产生一个输出的key/value pair 集合。MapReduce
库的用户用两个函数表达这个计算:Map 和Reduce。Hadoop Map/Reduce实现主要是通过继承Mapper和Reducer两个抽象类,并实现map和reduce两个方法实现的。 Mapper
Mapper 将输入键值对(key/value pair)映射到一组中间格式的键值对集合。 Map是一类将输入记录集转换为中间格式记录集的独立任务。 这种转换的中间格式记录集不需要与输入记录集的类型一致。一个给定的输入键值对可以映射成0个或多个输出键值对。
输出键值对不需要与输入键值对的类型一致。一个给定的输入键值对可以映射成0个或多个输出键值对。
Mapper 的输出被排序后,就被划分给每个Reducer 。分块的总数目和一个作业的reduce任务的数目是一样的。用户可以通过实现自定义的 Partitioner 来控制哪个key被分配给哪个 Reducer 。
用户可选择通过 JobConf.setCombinerClass(Class) 指定一个combiner ,它负责对中间过程的输出进行本地的聚集,这会有助于降低从Mapper 到 Reducer 数据传输量。
这些被排好序的中间过程的输出结果保存的格式是(key-len, key, value-len, value),应用程序可以通过JobConf 控制对这些中间结果是否进行压缩以及怎么压缩,使用哪种 CompressionCodec 。
需要多少个Map?
Map的数目通常是由输入数据的大小决定的,一般就是所有输入文件的总块(block)数。 Map正常的并行规模大致是每个节点(node)大约10到100个map,对于CPU 消耗较小的map任务可以设到300个左右。由于每个任务初始化需要一定的时间,因此,比较合理的情况是map执行的时间至少超过1分钟。
这样,如果你输入10TB的数据,每个块(block)的大小是128MB,你将需要大约82,000个map来完成任务,除非使用 setNumMapTasks(int) (注意:这里仅仅是对框架进行了一个提示(hint),实际决定因素见这里 )将这个数值设置得更高。
Reducer
Reducer 将与一个key关联的一组中间数值集归约(reduce)为一个更小的数值集。 用户可以通过 JobConf.setNumReduceTasks(int) 设定一个作业中reduce任务的数目。
概括地说,对Reducer 的实现者需要重写 JobConfigurable.configure(JobConf) 方法,这个方法需要传递一个JobConf 参数,目的是完成Reducer的初始化工作。然后,框架为成组的输入数据中的每个<key, (list of values)> 对调用一次 reduce(WritableComparable, Iterator, OutputCollector, Reporter) 方法。之后,应用程序可以通过重写Closeable.close() 来执行相应的清理工作。
Reducer 有3个主要阶段:shuffle、sort和reduce。
Shuffle Reducer 的输入就是Mapper已经排好序的输出。在这个阶段,框架通过HTTP为每个Reducer获得所有Mapper输出中与之相关的分块。 Sort 这个阶段,框架将按照key的值对Reducer 的输入进行分组 (因为不同mapper的输出中可能会有相同的key)。 Shuffle和Sort两个阶段是同时进行的;map的输出也是一边被取回一边被合并的。
需要多少个Reduce?
Reduce的数目建议是0.95 或1.75 乘以 (<no. of nodes > * mapred.tasktracker.reduce.tasks.maximum )。 用0.95,所有reduce可以在maps一完成时就立刻启动,开始传输map的输出结果。用1.75,速度快的节点可以在完成第一轮reduce任务后,可以开始第二轮,这样可以得到比较好的负载均衡的效果。
增加reduce的数目会增加整个框架的开销,但可以改善负载均衡,降低由于执行失败带来的负面影响。
上述比例因子比整体数目稍小一些是为了给框架中的推测性任务(speculative-tasks) 或失败的任务预留一些reduce的资源。
无Reducer 如果没有归约要进行,那么设置reduce任务的数目为零 是合法的。 这种情况下,map任务的输出会直接被写入由 setOutputPath(Path) 指定的输出路径。框架在把它们写入FileSystem 之前没有对它们进行排序。
知道了Map/Reduce相关基础知识,现在我们要做的事,就是对一个包含有海量数字的文本文件进行统计,并求出所有数字的和。
例子:对包含有1*10^6(100000)个数字文件,进行分析并求和。
文件格式:
-50 43 20 58 40 64 -95 28 61 55
38 78 -28 96 35 2 3 4 -87 22 -22 63 40 93 -58 81 72 63 93 94 -48 77 40 42 35 86 -66 43 26 70 -21 45 -14 6 21 73 96 31 -90 57 解决思路:
第一种方法是用Mapper读取文本文件用StringTokenizer对读取文件内的每一行的数字(Hadoop处理文本文件时,处理时是一行一行记取的)进行分隔,获取每一个数字,然后求和,再将求得的值按Key/Value格式写入Context,最后用Reducer对求得中间值进行汇总求和,得出整个文件所有数字的和。
第二种方法是用Mapper读取文本文件用StringTokenizer对文件内的数字进行分隔,获取每一个数字,并救出文件中该数字有多少个,在合并过程中,求出每个数字在文件中的和,最后用Reducer对求得每个数字求得的和进行汇总求和,得出整个文件所有数字的和。
下面就是实现的具体代码>>
第一种实现代码:
view plaincopy to clipboardprint?
01.package com.metarnet.hadoop; 02.import java.io.IOException; 03.import java.util.StringTokenizer; 04.import org.apache.hadoop.conf.Configuration; 05.import org.apache.hadoop.fs.Path; 06.import org.apache.hadoop.io.LongWritable; 07.import org.apache.hadoop.io.Text; 08.import org.apache.hadoop.mapreduce.Job; 09.import org.apache.hadoop.mapreduce.Mapper; 10.import org.apache.hadoop.mapreduce.Reducer; 11.import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 12.import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 13.import org.apache.hadoop.util.GenericOptionsParser; 14.public class NumberSum { 15. 16. 17. //对每一行数据进行分隔,并求和 18. public static class SumMapper extends 19. Mapper<Object, Text, Text, LongWritable> { 20. private Text word = new Text("sum"); 21. private static LongWritable numValue = new LongWritable(1); 22. public void map(Object key, Text value, Context context) 23. throws IOException, InterruptedException { 24. StringTokenizer itr = new StringTokenizer(value.toString()); 25. long sum = 0; 26. while (itr.hasMoreTokens()) { 27. String s = itr.nextToken(); 28. long val = Long.parseLong(s); 29. sum += val; 30. } 31. numValue.set(sum); 32. context.write(word, numValue); 33. } 34. } 35. 36. // 汇总求和,输出 37. public static class SumReducer extends 38. Reducer<Text, LongWritable, Text, LongWritable> { 39. private LongWritable result = new LongWritable(); 40. private Text k = new Text("sum"); 41. public void reduce(Text key, Iterable<LongWritable> values, 42. Context context) throws IOException, InterruptedException { 43. long sum = 0; 44. for (LongWritable val : values) { 45. long v = val.get(); 46. sum += v; 47. } 48. result.set(sum); 49. context.write(k, result); 50. } 51. } 52. /** 53. * @param args 54. * @throws Exception 55. */ 56. public static void main(String[] args) throws Exception { 57. Configuration conf = new Configuration(); 58. String[] otherArgs = new GenericOptionsParser(conf, args) 59. .getRemainingArgs(); 60. if (otherArgs.length != 2) { 61. System.err.println("Usage: numbersum <in> <out>"); 62. System.exit(2); 63. } 64. Job job = new Job(conf, "number sum"); 65. job.setJarByClass(NumberSum.class); 66. job.setMapperClass(SumMapper.class); 67. job.setReducerClass(SumReducer.class); 68. job.setOutputKeyClass(Text.class); 69. job.setOutputValueClass(LongWritable.class); 70. FileInputFormat.addInputPath(job, new Path(otherArgs[0])); 71. FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); 72. System.exit(job.waitForCompletion(true) ? 0 : 1); 73. System.out.println("ok"); 74. } 75.} 第二种实现代码: view plaincopy to clipboardprint?
01.package com.metarnet.hadoop; 02.import java.io.IOException; 03.import java.util.StringTokenizer; 04.import org.apache.hadoop.conf.Configuration; 05.import org.apache.hadoop.fs.Path; 06.import org.apache.hadoop.io.LongWritable; 07.import org.apache.hadoop.io.Text; 08.import org.apache.hadoop.mapreduce.Job; 09.import org.apache.hadoop.mapreduce.Mapper; 10.import org.apache.hadoop.mapreduce.Reducer; 11.import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 12.import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 13.import org.apache.hadoop.util.GenericOptionsParser; 14.public class NumberSum { 15. 16. //对每一个数字进行分隔 17. public static class NumSumMapper extends 18. Mapper<Object, Text, Text, LongWritable> { 19. private Text word = new Text(); 20. private static LongWritable numValue = new LongWritable(1); 21. public void map(Object key, Text value, Context context) 22. throws IOException, InterruptedException { 23. StringTokenizer itr = new StringTokenizer(value.toString()); 24. long sum = 0; 25. while (itr.hasMoreTokens()) { 26. String s = itr.nextToken(); 27. word.set(s); 28. context.write(word, numValue); 29. } 30. } 31. } 32. 33. //对每一个数字进行汇总求和 34. public static class SumCombiner extends 35. Reducer<Text, LongWritable, Text, LongWritable> { 36. private LongWritable result = new LongWritable(); 37. private Text k = new Text("midsum"); 38. public void reduce(Text key, Iterable<LongWritable> values, 39. Context context) throws IOException, InterruptedException { 40. long sum = 0; 41. if (!key.toString().startsWith("midsum")) { 42. for (LongWritable val : values) { 43. sum += val.get(); 44. } 45. long kval = Long.parseLong(key.toString()); 46. long v = kval * sum; 47. result.set(v); 48. context.write(k, result); 49. } else { 50. for (LongWritable val : values) { 51. context.write(key, val); 52. } 53. } 54. } 55. } 56. // 汇总求和,输出 57. public static class SumReducer extends 58. Reducer<Text, LongWritable, Text, LongWritable> { 59. private LongWritable result = new LongWritable(); 60. private Text k = new Text("sum"); 61. public void reduce(Text key, Iterable<LongWritable> values, 62. Context context) throws IOException, InterruptedException { 63. long sum = 0; 64. for (LongWritable val : values) { 65. long v = val.get(); 66. sum += v; 67. } 68. result.set(sum); 69. context.write(k, result); 70. } 71. } 72. /** 73. * @param args 74. * @throws Exception 75. */ 76. public static void main(String[] args) throws Exception { 77. Configuration conf = new Configuration(); 78. String[] otherArgs = new GenericOptionsParser(conf, args) 79. .getRemainingArgs(); 80. if (otherArgs.length != 2) { 81. System.err.println("Usage: numbersum <in> <out>"); 82. System.exit(2); 83. } 84. Job job = new Job(conf, "number sum"); 85. job.setJarByClass(NumberSum.class); 86. job.setMapperClass(NumSumMapper.class); 87. job.setCombinerClass(SumCombiner.class); 88. job.setReducerClass(SumReducer.class); 89. job.setOutputKeyClass(Text.class); 90. job.setOutputValueClass(LongWritable.class); 91. FileInputFormat.addInputPath(job, new Path(otherArgs[0])); 92. FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); 93. System.exit(job.waitForCompletion(true) ? 0 : 1); 94. System.out.println("ok"); 95. } 96.} 第一种实现的方法相对简单,第二种实现方法用到了Combiner类,Hadoop 在 对中间求的的数值进行Combiner时,是通过递归的方式不停地对 Combiner方法进行调用,进而合并数据的。
从两种方法中,我们可以看出Map/Reduce的核心是在怎样对输入的数据按照何种方式是进行Key/Value分对的,分的合理对整个结果输出及算法实现有很大的影响。
本文来自CSDN博客,转载请标明出处:http://blog.csdn.net/savechina/archive/2010/08/24/5835933.aspx |
|