图文精华

About云-梭伦科技»专题 › 交流区 › 技术交流 › MapReduce › MapReduce不执行Reduce
分享

MapReduce不执行Reduce

Joker 发表于 2014-10-19 14:43:58 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 5 24243 

  1. package cn.basemapreduce;

  2. 

  3. import java.io.IOException;

  4. import java.util.StringTokenizer;

  5. 

  6. import org.apache.hadoop.conf.Configuration;

  7. import org.apache.hadoop.fs.Path;

  8. import org.apache.hadoop.io.IntWritable;

  9. import org.apache.hadoop.io.LongWritable;

  10. import org.apache.hadoop.io.Text;

  11. import org.apache.hadoop.mapreduce.Job;

  12. import org.apache.hadoop.mapreduce.Mapper;

  13. import org.apache.hadoop.mapreduce.Reducer;

  14. import org.apache.hadoop.mapreduce.Reducer.Context;

  15. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

  16. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

  17. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

  18. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

  19. import org.apache.hadoop.util.GenericOptionsParser;

  20. 

  21. 

  22. 

  23. public class Filter3 {

  24. 

  25.         public static class Map extends

  26.                         Mapper<LongWritable, Text, Text, LongWritable> {

  27. 

  28.                 @Override

  29.                 protected void map(LongWritable key, Text value,

  30.                                 org.apache.hadoop.mapreduce.Mapper.Context context)

  31.                                 throws IOException, InterruptedException {

  32.                         // TODO Auto-generated method stub

  33. 

  34.                         String line = value.toString();

  35. 

  36.                         StringTokenizer tokenizer = new StringTokenizer(line, "\n");

  37. 

  38.                         while (tokenizer.hasMoreElements()) {

  39. 

  40.                                 StringTokenizer st = new StringTokenizer(tokenizer.nextToken());

  41. 

  42.                                 String score = st.nextToken();

  43.                                 String name = st.nextToken();

  44.                                 

  45.                                 if(name.equals("赵六")){

  46.                                         

  47.                                         int tscore = Integer.parseInt(score);

  48.                                         Text tname = new Text(name);

  49.                                         

  50.                                         context.write(name, new LongWritable(tscore));

  51.                                 }

  52.                                 

  53.                         }

  54.                 }

  55. 

  56.         }

  57.         

  58.         

  59.         public static class Reduce extends Reducer<Text, LongWritable, Text, Text>{

  60.                 

  61.                 

  62.                 protected void reduce(Text key, Iterable<LongWritable> values,Context context)

  63.                                 throws IOException, InterruptedException {

  64.                         // TODO Auto-generated method stub

  65.                         context.write(key, new Text(""));

  66.                 }

  67.                 

  68.         }

  69.         

  70.         public static void main(String[] args) throws Exception {

  71.                 

  72.                 Configuration conf = new Configuration();

  73.                 // conf.set("mapred.job.tracker", "192.168.1.2:9001");

  74. 

  75.                 String[] ioArgs = new String[] { "mkdir", "score_out" };

  76. 

  77.                 String[] otherArgs = new GenericOptionsParser(conf, ioArgs)

  78.                                 .getRemainingArgs();

  79. 

  80.                 if (otherArgs.length != 2) {

  81.                         System.err.println("Usage: Score Average <in> <out>");

  82. 

  83.                         System.exit(2);

  84.                 }

  85. 

  86.                 Job job = new Job(conf, "Filter3");

  87. 

  88.                 job.setJarByClass(Filter3.class);

  89. 

  90.                 // 设置Map、Combine和Reduce处理类

  91.                 job.setMapperClass(Map.class);

  92.                 //它负责对中间过程的输出进行本地的聚集，这会有助于降低从Mapper到 Reducer数据传输量

  93.                 job.setCombinerClass(Reduce.class);

  94.                 job.setReducerClass(Reduce.class);

  95. 

  96.                 // 设置输出类型

  97.                 job.setOutputKeyClass(Text.class);

  98.                 job.setOutputValueClass(Text.class);

  99. 

  100.                 // 将输入的数据集分割成小数据块splites，提供一个RecordReder的实现

  101.                 job.setInputFormatClass(TextInputFormat.class);

  102. 

  103.                 // 提供一个RecordWriter的实现，负责数据输出

  104.                 job.setOutputFormatClass(TextOutputFormat.class);

  105. 

  106.                 // 设置输入和输出目录

  107.                 FileInputFormat.addInputPath(job, new Path(otherArgs[0]));

  108.                 FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));

  109. 

  110.                 System.exit(job.waitForCompletion(true) ? 0 : 1);

  111.         }

  112. 

  113. }

复制代码
自己写的时候Reduce总是不会被执行，不知道哪里有问题?
大家帮忙看下



回复

使用道具 举报

自定义广告语

已有(5)人评论

跳转到指定楼层
正序浏览
desehawk 发表于 2014-10-19 15:49:10
hadoop1还是hadoop2，是在集群环境，还是eclipse环境。

  protected void reduce(Text key, Iterable<LongWritable> values,Context context)
                                throws IOException, InterruptedException {
                        // TODO Auto-generated method stub
                        context.write(key, new Text(""));
                }



上面红字部分替换为下面：

  1. ctx.write(key, new LongWritable(1));
复制代码



回复

使用道具 举报

Joker 发表于 2014-10-19 15:53:24
desehawk 发表于 2014-10-19 15:49
hadoop1还是hadoop2，是在集群环境，还是eclipse环境。

  protected void reduce(Text key, Iterable va ...

Hadoop2.2我是在eclipse环境下

我Reduce定义的是Text,改成IntWritable报错不行
是不是我要先改下类型
回复

使用道具 举报

howtodown 发表于 2014-10-19 16:04:58
Joker 发表于 2014-10-19 15:53
Hadoop2.2我是在eclipse环境下

我Reduce定义的是Text,改成IntWritable报错不行
hadoop2.2已经是新api了，感觉你用的还是旧 api.试试下面程序，直接复制你的程序，然后准备数据运行。
  1. package mapreduce;

  2. 

  3. import java.io.IOException;

  4. 

  5. import org.apache.hadoop.conf.Configuration;

  6. import org.apache.hadoop.fs.Path;

  7. import org.apache.hadoop.io.LongWritable;

  8. import org.apache.hadoop.io.Text;

  9. import org.apache.hadoop.mapreduce.Job;

  10. import org.apache.hadoop.mapreduce.Mapper;

  11. import org.apache.hadoop.mapreduce.Reducer;

  12. import org.apache.hadoop.mapreduce.lib.output.*;

  13. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

  14. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

  15. import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;

  16. 

  17. public class mapreduce {

  18. 

  19. static final String INPUT_PATH = "hdfs://aboutyun:9000/yy1";

  20. static final String OUT_PATH = "hdfs://aboutyun:9000/Output";

  21. 

  22. public static void main(String[] args) throws Exception {

  23. //主类

  24. Configuration conf = new Configuration();

  25. final Job job = new Job(conf, mapreduce.class.getSimpleName());

  26. job.setJarByClass(mapreduce.class);

  27. // 寻找输入

  28. FileInputFormat.setInputPaths(job, INPUT_PATH);

  29. // 1.2对输入数据进行格式化处理的类

  30. job.setInputFormatClass(TextInputFormat.class);

  31. job.setMapperClass(MyMapper.class);

  32. 

  33. // 1.2指定map输出类型<key,value>类型

  34. job.setMapOutputKeyClass(Text.class);

  35. job.setMapOutputValueClass(LongWritable.class);

  36. 

  37. // 1.3指定分区

  38. job.setPartitionerClass(HashPartitioner.class);

  39. job.setNumReduceTasks(1);

  40. 

  41. // 1.4排序分组省略，使用默认

  42. // 1.5规约省略，使用默认

  43. job.setReducerClass(MyReduce.class);

  44. job.setOutputKeyClass(Text.class);

  45. job.setOutputValueClass(LongWritable.class);

  46. // 指定输出路径

  47. FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));

  48. // 指定输出的格式或则类

  49. job.setOutputFormatClass(TextOutputFormat.class);

  50. 

  51. // 把作业提交给jobtracer

  52. job.waitForCompletion(true);

  53. 

  54. }

  55. //map类

  56. static class MyMapper extends

  57. Mapper<LongWritable, Text, Text, LongWritable> {

  58. protected void map(LongWritable key, Text value, Context context)

  59. throws IOException, InterruptedException {

  60. final String[] splited = value.toString().split("\t");

  61. for (String word : splited) {

  62. context.write(new Text(word), new LongWritable(1L));

  63. 

  64. }

  65. 

  66. }

  67. 

  68. }

  69. //reduce类

  70. static class MyReduce extends

  71. Reducer<Text, LongWritable, Text, LongWritable> {

  72. @Override

  73. protected void reduce(Text k2, java.lang.Iterable<LongWritable> v2s,

  74. Context ctx) throws java.io.IOException, InterruptedException {

  75. long times = 0L;

  76. for (LongWritable count : v2s) {

  77. times += count.get();

  78. ctx.write(k2, new LongWritable(times));

  79. }

  80. 

  81. }

  82. 

  83. }

  84. 

  85. }
复制代码

更多参考：
新手指导，该如何在开发环境中，创建mapreduce程序




回复

使用道具 举报

Joker 发表于 2014-10-19 16:20:21
howtodown 发表于 2014-10-19 16:04
hadoop2.2已经是新api了，感觉你用的还是旧 api.试试下面程序，直接复制你的程序，然后准备数据运行。

...

多谢，前些天看新手到工作中的编程篇，然后改变下写的
回复

使用道具 举报

evababy 发表于 2014-11-28 10:51:42
ctx.write(key, new LongWritable(1));job.setOutputValueClass(LongWritable.class);

回复

使用道具 举报

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
Joker

高级会员

关注

  • 44

    主题

  • 271

    帖子

  • 3

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2025 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000