载入中。。。 'S bLog
 
载入中。。。
  
载入中。。。
载入中。。。
载入中。。。
载入中。。。
载入中。。。
 
填写您的邮件地址，订阅我们的精彩内容：


 
《Hadoop in Action》第四章习题
[ 2012/2/26 22:07:00 | By: 梦翔儿 ]
 

《Hadoop in Action》第四章习题：

0.MaxValue:要求输出cite75_99.txt中最大的CITED值：

要点：

1.Mapper只输出它所处理的数据中的最大值。（重写cleanup()函数）

2.设置Reducer数目为一个 -D mapred.reduce.tasks=1，同时也只输出所处理的最大值。（重写cleanup()函数）

3.cleanup()函数：在任务结束时执行一次。详见API。

代码如下：

[java] view plaincopyprint?
  1. /*
  2. * MaxValues
  3. * 函数作用：输出Patent中最大数值
  4. * Author: jokes000
  5. * Date: 2011-12-15
  6. */
  8. import java.io.IOException;
  10. import org.apache.hadoop.conf.Configuration;
  11. import org.apache.hadoop.conf.Configured;
  12. import org.apache.hadoop.fs.Path;
  13. import org.apache.hadoop.io.IntWritable;
  14. import org.apache.hadoop.io.LongWritable;
  15. import org.apache.hadoop.io.Text;
  16. import org.apache.hadoop.mapreduce.Job;
  17. import org.apache.hadoop.mapreduce.Mapper;
  18. import org.apache.hadoop.mapreduce.Reducer;
  19. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  20. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  21. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  22. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  23. import org.apache.hadoop.util.Tool;
  24. import org.apache.hadoop.util.ToolRunner;
  27. public class MaxValue extends Configured implements Tool {
  29. public static class MapClass extends Mapper<LongWritable,Text,Text,Text> {
  31. int max = 0;
  32. // Map Method
  33. public void map(LongWritable key, Text value, Context context){
  34. String[] citation = value.toString().split(",", 2);
  35. try {
  36. int tmp = Integer.parseInt(citation[0]);
  37. if( tmp > max ) max = tmp;
  38. } catch(NumberFormatException e){
  39. // do nothing.
  40. }
  41. //context.write(new Text(citation[0]), new Text(citation[0]));
  42. }
  44. @Override
  45. protected void cleanup(Context context) throws IOException, InterruptedException {
  46. context.write(new Text(max+""), new Text(max+""));
  47. }
  48. }
  50. public static class Reduce extends Reducer<Text,Text,Text,IntWritable> {
  52. int max = 0;
  53. // Reduce Method
  54. public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
  55. //IntWritable[] top = new IntWritable[10];
  56. for(Text value : values) {
  57. try {
  58. int tmp = Integer.parseInt(value.toString());
  59. if( tmp > max ) max = tmp;
  60. } catch(NumberFormatException e) {
  61. // do nothing.
  62. }
  63. }
  64. //context.write(new Text("Max"), new IntWritable(max));
  65. }
  67. @Override
  68. protected void cleanup(Context context) throws IOException, InterruptedException {
  69. context.write(new Text("Max"), new IntWritable(max));
  70. }
  71. }
  73. @Override
  74. public int run(String[] arg0) throws Exception {
  76. Job job = new Job();
  77. job.setJarByClass(MaxValue.class);
  79. FileInputFormat.addInputPath(job, new Path(arg0[0]));
  80. FileOutputFormat.setOutputPath(job, new Path(arg0[1]));
  82. job.setMapperClass(MapClass.class);
  83. job.setReducerClass(Reduce.class);
  84. job.setInputFormatClass(TextInputFormat.class);
  85. job.setOutputFormatClass(TextOutputFormat.class);
  86. job.setOutputKeyClass(Text.class);
  87. job.setOutputValueClass(Text.class);
  89. job.waitForCompletion(true);
  91. return 0;
  92. }
  94. public static void main(String[] args) throws Exception {
  95. int res = ToolRunner.run(new Configuration(), new MaxValue(), args);
  96. System.exit(res);
  97. }
  99. }


1.Top K Values: 要求输出adapt63_99.txt中的第9列CLAIMS值的最大的K个值：

要点：

1.Mapper值输出它所处理的数据中的最大的K个值。(重写 cleanup()函数)

2.设置Reducer数目为1 -D mapred.reduce.tasks=1，同时对Mapper中输出进行排序，输出最大的K个值(重写 cleanup()函数)

代码如下：

[java] view plaincopyprint?
  1. /*
  2. * TopKValues
  3. * 函数作用：输出CLAIMS中最大的几个数值
  4. * Author: jokes000
  5. * Date: 2011-12-15
  6. */
  8. import java.io.IOException;
  9. import java.util.Arrays;
  11. import org.apache.hadoop.conf.Configuration;
  12. import org.apache.hadoop.conf.Configured;
  13. import org.apache.hadoop.fs.Path;
  14. import org.apache.hadoop.io.IntWritable;
  15. import org.apache.hadoop.io.LongWritable;
  16. import org.apache.hadoop.io.Text;
  17. import org.apache.hadoop.mapreduce.Job;
  18. import org.apache.hadoop.mapreduce.Mapper;
  19. import org.apache.hadoop.mapreduce.Reducer;
  20. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  21. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  22. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  23. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  24. import org.apache.hadoop.util.Tool;
  25. import org.apache.hadoop.util.ToolRunner;
  28. public class TopKValues extends Configured implements Tool {
  30. public static class MapClass extends Mapper<LongWritable,Text,Text,IntWritable> {
  31. // 全局变量
  32. int len; // K值
  33. int[] top; // 用于保存的数组
  35. // Map Method
  36. public void map(LongWritable key, Text value, Context context) {
  37. String[] fields = value.toString().split(",",-20);
  38. try {
  39. int claims = Integer.parseInt(fields[8]);
  40. add(claims);
  41. } catch(NumberFormatException e) {
  42. // do nothing..
  43. }
  44. }
  46. private void add(int value) {
  47. top[0] = value;
  48. Arrays.sort(top);
  49. }
  51. @Override
  52. protected void setup(Context context) {
  53. // 获取设置的"K"值，若没有K值，则设置该值为10
  54. len = context.getConfiguration().getInt("K", 10);
  55. top = new int[len+1];
  56. }
  58. @Override
  59. protected void cleanup(Context context) throws IOException, InterruptedException {
  60. for( int i = 1; i <= len; ++ i ) {
  61. context.write(new Text(top[i]+""), new IntWritable(top[i]));
  62. }
  63. }
  64. }
  66. public static class Reduce extends Reducer<Text,IntWritable,Text,IntWritable> {
  67. int[] top;
  68. int len;
  70. @Override
  71. protected void setup(Context context) {
  72. len = context.getConfiguration().getInt("K", 10);
  73. top = new int[len+1];
  74. }
  76. private void add(int value) {
  77. top[0] = value;
  78. Arrays.sort(top);
  79. }
  81. // Reduce Method
  82. public void reduce(Text key, Iterable<IntWritable> values, Context context) {
  83. for(IntWritable value : values) {
  84. add(value.get());
  85. }
  86. }
  88. @Override
  89. protected void cleanup(Context context) throws IOException, InterruptedException {
  90. for( int i = len; i > 0; -- i ) {
  91. context.write(new Text("No."+(len-i+1)), new IntWritable(top[i]));
  92. }
  93. }
  94. }
  96. @Override
  97. public int run(String[] arg0) throws Exception {
  99. Job job = new Job();
  100. job.setJarByClass(TopKValues.class);
  102. FileInputFormat.addInputPath(job, new Path(arg0[0]));
  103. FileOutputFormat.setOutputPath(job, new Path(arg0[1]));
  105. try{
  106. int K = Integer.parseInt(arg0[2]);
  107. getConf().setInt("K", K);
  108. } catch(NumberFormatException e) {
  109. // do nothing..
  110. getConf().setInt("K", 20);
  111. }
  113. job.setMapperClass(MapClass.class);
  114. job.setReducerClass(Reduce.class);
  115. job.setInputFormatClass(TextInputFormat.class);
  116. job.setOutputFormatClass(TextOutputFormat.class);
  117. job.setOutputKeyClass(Text.class);
  118. job.setOutputValueClass(IntWritable.class);
  120. job.waitForCompletion(true);
  122. return 0;
  123. }
  125. public static void main(String[] args) throws Exception {
  126. int res = ToolRunner.run(new Configuration(), new TopKValues(), args);
  127. System.exit(res);
  128. }
  130. }

http://blog.csdn.net/jokes000/article/details/7075344

 
 
  • 标签：Hadoop 
    • 发表评论：
    载入中。。。

    		  
    		  
     

    梦翔儿网站 梦飞翔的地方 http://www.dreamflier.net
    中华人民共和国信息产业部TCP/IP系统 备案序号：辽ICP备09000550号

    Powered by Oblog.